#include "market.h"
#include "ctp.h"
#include <mdbdf.hpp>
#include <zqstr.hpp>
#include <zqpb.hpp>
#include <zqcsv.hpp>
#include <zqjson.hpp>

CTPMarket::CTPMarket(boost::asio::io_service& service, std::vector<HZQDB>& subs)
	: service_(service)
	//, timer_(service)
	, subs_(subs)
{

}

bool CTPMarket::init(const CTPUserLoginInfo& userinfo, size_t pos)
{
	userinfo_ = userinfo;
	srv_pos_ = pos;
	userid_ = userinfo_.UserID + std::string(".") + userinfo_.BrokerID;
	return true;
}

void CTPMarket::start()
{
	if (ZQDBIsTest()) {
	}
	else {
		//service().post(boost::bind(&This::set_online, this));
	}
}

void CTPMarket::stop()
{
	stop_flag_ = true;
	//unregister_timer();
	if (ZQDBIsTest()) {
	}
	else {
		service().post(boost::bind(&This::set_offline, this));
	}
}

/*void CTPMarket::register_timer()
{
LOG4I("CTPMarket::register_timer");
timer_.expires_from_now(boost::posix_time::milliseconds(60 * 1000));
timer_.async_wait(boost::bind(&This::on_timer, this, boost::asio::placeholders::error));
}

void CTPMarket::unregister_timer()
{
LOG4I("CTPMarket::unregister_timer");
boost::system::error_code ec;
timer_.cancel(ec);
}*/

//以下是当前service线程回调函数

void CTPMarket::set_online()
{
	ASSERT(!api_);
	LOG4I(STR_MODULE_NAME " market %s connecting", userid_.c_str());
	MDMgr::Inst().set_offline(shared_from_this());

	boost::filesystem::path flowpath(ZQDBGetDataDir());flowpath.append(".my");
	flowpath.append(STR_MODULE_NAME).append(userinfo_.BrokerID).append(userinfo_.UserID).append(std::to_string(srv_pos_));
	if (!boost::filesystem::exists(flowpath))
	{
		boost::filesystem::create_directories(flowpath);
	}
	//boost::filesystem::path currentpath = boost::filesystem::current_path();
	//boost::filesystem::current_path(flowpath);
	api_ = CThostFtdcMdApi::CreateFtdcMdApi(flowpath.string().c_str());
	//boost::filesystem::current_path(currentpath);
	//for (size_t i = 0; i<userinfo_.MdFrontAddress.size(); i++)
	{
		api_->RegisterFront((char*)userinfo_.MdFrontAddress[srv_pos_].c_str());
	}
	/*for (size_t i = 0; i<userinfo_.MdNsAddress.size(); i++)
	{
		api_->RegisterNameServer((char*)userinfo_.MdNsAddress[i].c_str());
	}*/

	api_->RegisterSpi(this);

	// 使客户端开始与行情发布服务器建立连接
	api_->Init();
}

void CTPMarket::set_offline()
{
	LOG4I(STR_MODULE_NAME " market %s disconnecting", userid_.c_str());
	normal_flag_ = false;
	MDMgr::Inst().set_offline(shared_from_this());
	if (is_stop()) {

	}
	if (api_) {
		api_->Release();
		api_ = NULL;
	}
}

void CTPMarket::subscribe(const std::vector<HZQDB>& subs)
{
	service().post(boost::bind(&This::on_subscribe, this, subs));
}

void CTPMarket::unsubscribe(const std::vector<HZQDB>& subs)
{
	service().post(boost::bind(&This::on_unsubscribe, this, subs));
}

void CTPMarket::on_subscribe(std::vector<HZQDB>& subs)
{
	if (is_stop()) {
		return;
	}

	if (!api_) {
		set_online();
	}
	else {
		if (normal_flag_ && !subs.empty()) {
			do_subscribe(subs);
		}
	}
}

void CTPMarket::on_unsubscribe(std::vector<HZQDB>& subs)
{
	if (is_stop()) {
		return;
	}
	if (subs.empty()) {
		if (api_) {
			set_offline();
		}
	}
	else {
		if (normal_flag_ && !subs.empty()) {
			do_unsubscribe(subs);
		}
	}
}

/*void CTPMarket::on_timer(const boost::system::error_code& ec)
{
if (ec) {
return;
}

if (!api_) {
return;
}
if (!normal_flag_) {
set_offline();
set_online();
}
else {
//
}
}*/

void CTPMarket::on_connect()
{
	LOG4I(STR_MODULE_NAME " market %s connect", userid_.c_str());
	if (userinfo_.UserID[0]) {
		api_->ReqUserLogin((CThostFtdcReqUserLoginField*)&userinfo_, 0);
	}
	else {
		on_normal();
	}
}

void CTPMarket::on_disconnect()
{
	LOG4I(STR_MODULE_NAME " market %s disconnect", userid_.c_str());
	normal_flag_ = false;
	MDMgr::Inst().set_offline(shared_from_this());
}

void CTPMarket::on_login()
{
	on_normal();
}

void CTPMarket::on_normal()
{
	LOGXXX("CTPMarket on_normal");
	normal_flag_ = true;
	MDMgr::Inst().set_online(shared_from_this());
	//重新订阅
	do_subscribe(subs_);
}

void CTPMarket::do_subscribe(const std::vector<HZQDB>& subs)
{
	std::set<const char*, XUtil::strless> codeset;
	for (auto h : subs)
	{
		zqdb::Code code(h);
		codeset.insert(code->TradeCode);
	}
	codes_.clear();
	for (auto code : codeset)
	{
		codes_.push_back(code);
	}
	/*if (!codes_.empty()) {
		api_->SubscribeMarketData((char**)&codes[0], codes.size());
	}*/
	code_next_ = 0;
	do_subscribe_next();
}

void CTPMarket::do_unsubscribe(const std::vector<HZQDB>& subs)
{
	std::set<const char*, XUtil::strless> codeset;
	for (auto h : subs)
	{
		zqdb::Code code(h);
		codeset.insert(code->TradeCode);
	}
	std::vector<const char*> codes;
	for (auto code : codeset)
	{
		codes.push_back(code);
	}
	if (!codes.empty()) {
		api_->UnSubscribeMarketData((char**)&codes[0], codes.size());
	}
}

void CTPMarket::do_subscribe_next()
{
	LOGXXX("CTPMarket do_subscribe_next=%d", code_next_);
	if (code_next_ < codes_.size()) {
		api_->SubscribeMarketData((char**)&codes_[code_next_], std::min<>((size_t)20, codes_.size() - code_next_));
		code_next_ += 20;
	}
}

void CTPMarket::on_rsp_subscribe_next(const CThostFtdcSpecificInstrumentField& specificInstrument, bool bIsLast)
{
	if (bIsLast) {
		do_subscribe_next();
	}
}

void CTPMarket::on_market_data_udpate(const CThostFtdcDepthMarketDataField& data)
{
	if (is_stop()) {
		return;
	}
	if (!normal_flag_) {
		return;
	}
	
	MDMgr::Inst().update_data(data);
}

//protected:
//以下是来自行情回调

void CTPMarket::OnFrontConnected()
{
	service().post(boost::bind(&This::on_connect, this));
}

void CTPMarket::OnFrontDisconnected(int nReason)
{
	service().post(boost::bind(&This::on_disconnect, this));
}

void CTPMarket::OnRspUserLogin(CThostFtdcRspUserLoginField *pRspUserLogin, CThostFtdcRspInfoField *pRspInfo, int nRequestID, bool bIsLast)
{
	LOG4I(STR_MODULE_NAME " market %s login", userid_.c_str());
	service().post(boost::bind(&This::on_login, this));
}

void CTPMarket::OnRspSubMarketData(CThostFtdcSpecificInstrumentField *pSpecificInstrument, CThostFtdcRspInfoField *pRspInfo, int nRequestID, bool bIsLast)
{
	if (IsErrorRspInfo(pRspInfo)) {
		LOG4E(STR_MODULE_NAME " %s OnRspSubMarketData %s", userid_.c_str(), gbk_to_utf8(pRspInfo->ErrorMsg).c_str());
	}
	service().post(boost::bind(&This::on_rsp_subscribe_next, this, pSpecificInstrument ? *pSpecificInstrument : CThostFtdcSpecificInstrumentField{}, bIsLast));
}

void CTPMarket::OnRtnDepthMarketData(CThostFtdcDepthMarketDataField *pDepthMarketData)
{
	if (pDepthMarketData) {
		service().post(boost::bind(&This::on_market_data_udpate, this, *pDepthMarketData));
		//Manager::Inst().update_data(*pDepthMarketData);
	}
}

///

MDMgr::MDMgr(): Base()
, timer_(service())
{

}

//主力合约代码是昨持仓最大的合约
//商品期货的编码规则:“品种” + 合约到期年份 + 合约到期月份；

//股指期货的代码规则：如沪深300股指期货，IF+合约到期年份+合约到期月份，
//国债期货合约规则：为最近3个季月（3 / 6 / 9 / 12中的最近3个），如5年期国债期货TF，TF + 合约到期年份 + 合约到期月份

//这里我们所有期货合约去掉年份，合约数全都转成12个，增加主力合约0000，并默认根据实际合约数增加对应连续合约（0001-0012）,这样默认主力和连续合约排序在前
//例如IF
//IF0000 IF0001 ... IF0012 IF01 ... IF12 

//由于部分期货存在12期的合约，比如bu存在bu2106 bu2206，所以我们这里连续合约使用61+合约idx生成代码

//这里调整下只支持主力XX0，次主力XX00，标准合约XX0YMM

bool MDMgr::IsInstrumentMain(const std::shared_ptr<CTPCommodityInfoEx> Instrument) {
	return Instrument->product_->main_commodity_ == Instrument;
}
bool MDMgr::IsInstrumentSecond(const std::shared_ptr<CTPCommodityInfoEx> Instrument) {
	return Instrument->product_->second_commodity_ == Instrument;
}
const char* MDMgr::Instrument2Code(uint32_t tradingday, const std::shared_ptr<CTPCommodityInfoEx> Instrument, char* Code)
{
	int ProductLen = strlen(Instrument->Product);
	int CodeLen = strlen(Instrument->Code);
	if (CodeLen > (ProductLen + 4)) {
		ASSERT(0);
	}
	else if (CodeLen == (ProductLen + 4)) {
		uint32_t Year, Month, Day;
		XUtil::SplitDate(tradingday, Year, Month, Day);
		Year %= 100;
		int CodeYYMM = std::atoi(&Instrument->Code[CodeLen - 4]);
		int CodeYear = CodeYYMM / 100;
		int CodeMonth = CodeYYMM % 100;
		if (CodeYear > (Year + 1)) {
			sprintf(Code, "%s%02d%s", Instrument->Product, CodeYear - Year, &Instrument->Code[CodeLen - 2]);
		}
		else if (CodeYear > Year && CodeMonth >= Month) {
			sprintf(Code, "%s%02d%s", Instrument->Product, CodeYear - Year, &Instrument->Code[CodeLen - 2]);
		}
		else {
			sprintf(Code, "%s00%s", Instrument->Product, &Instrument->Code[CodeLen - 2]);
		}
	}
	else {
		uint32_t Year, Month, Day;
		XUtil::SplitDate(tradingday, Year, Month, Day);
		sprintf(Code, "%s%02d%s", Instrument->Product, (Instrument->pos + Month) / 12, &Instrument->Code[CodeLen - 2]);
	}
	return Code;
}
const char* MDMgr::Instrument2Mian(const std::shared_ptr<CTPCommodityInfoEx> Instrument, char* Code)
{
	sprintf(Code, "%s0", Instrument->Product);
	return Code;
}
const char* MDMgr::Instrument2Second(const std::shared_ptr<CTPCommodityInfoEx> Instrument, char* Code)
{
	sprintf(Code, "%s00", Instrument->Product);
	return Code;
}

void MDMgr::init()
{
	auto& cfg = Manager::Inst().cfg();
	sub_all_ = cfg.get<int>("subscribe_all", sub_all_);
	//MaxProductCountPerExchange_ = cfg.get<size_t>("max_product_per_exchange", 100);
	//MaxCodeCountPerProduct_ = cfg.get<size_t>("max_code_per_product", 100);
	auto max_data_reserved = cfg.get<size_t>("max_data_reserved", 0);
	for (size_t i = 0; i < CYC_MAX; i++)
	{
		char buf[260] = { 0 };
		sprintf(buf, "max_%s_reserved", ZQDBCycle2Str((PERIODTYPE)i));
		auto opt_value = cfg.get_optional<size_t>(buf);
		if (opt_value) {
			MaxDataReservedCount_[i] = opt_value.get();
		} else {
			MaxDataReservedCount_[i] = max_data_reserved;
		}
	}
	holiday_ = ZQDBFindHoliday(EXCHANGE_CFFEX);
	ASSERT(holiday_);
}

bool MDMgr::start()
{
	if (ZQDBIsTest()) {
		tradingday_ = g_module_info->tradingday;
		ASSERT(tradingday_);
		init_base_data();
	}
	else {
		boost::system::error_code ec;
		auto dara_dir = ZQDBGetDataDir();
		if (dara_dir && dara_dir[0]) {
			char buf[1024] = { 0 };
			snprintf(buf, 1024, "%s/%s/market", dara_dir, STR_MODULE_NAME);
			if (!boost::filesystem::exists(buf, ec)) {
				boost::filesystem::create_directories(buf, ec);
			}
		}
		auto& cfg = Manager::Inst().cfg();
		auto opt_user = cfg.get_child_optional("user");
		if (opt_user)
		{
			const boost::property_tree::ptree &cfg_user = opt_user.get();
			BOOST_FOREACH(const boost::property_tree::ptree::value_type &cfgi, cfg_user)
			{
				const auto& cfg_one = cfgi.second;
				std::string user = cfg_one.get<std::string>("UserID", "");
				{
					CTPUserLoginInfo userinfo;
					if (parse_userinfo(userinfo, cfg_one, true)) {
						for (size_t i = 0; i < userinfo.MdFrontAddress.size(); i++)
						{
							auto market = std::make_shared<CTPMarket>(service(), subs_);
							if (market->init(userinfo, i)) {
								markets_.push_back(market);
							}
						}
					}
				}
			}
		}
		if (markets_.empty()) {
			//表示使用官方行情
			g_module_info->tradingday = 0;
			g_module_info->closeday = 0;
		}
		else {
			g_module_info->tradingday = 0;
			for (auto Exchange : all_exchange)
			{
				auto h = ZQDBGetExchange(Exchange);
				if (h) {
					zqdb::Exchange exchange(h);
					g_module_info->tradingday = exchange->TradeDay;
					break;
				}
			}
			tradingday_ = g_module_info->tradingday;
			if (!g_module_info->tradingday) {
				g_module_info->closeday = 0;
			}
		}
	}
	Base::start();
	return true;
}

void MDMgr::run()
{
	if (ZQDBIsTest()) {
		//启动时自己normal，
		//后面通过begin/init/end_newtradingday完成交易日切换，
		//实盘也是通过begin/init/end_newtradingday完成交易日切换。
		service().post(boost::bind(&MDMgr::on_normal, this, false));
	}
	else {
		if (markets_.empty()) {
			//表示使用官方行情
			ASSERT(!online_flag_);
		}
		else {
			for (auto& market : markets_) {
				market->start();
			}
			//trader会触发on_normal
		}
	}
}

void MDMgr::stop()
{
	stop_flag_ = true;

	unregister_timer();

	{
		//std::unique_lock<std::mutex> lock(mutex_);

		for (auto& market : markets_)
		{
			market->stop();
		}
	}

	Base::stop();

	markets_.clear();

	close_record();
}

void MDMgr::on_new_tradingday2(uint32_t trading_day)
{
	LOG4I(STR_MODULE_NAME " market new tradingday2=%d init_flag=%d", (int)trading_day, (int)init_flag_);
	if (init_flag_) {
		return;
	}
	init_flag_ = true;
	unregister_timer();
	tradingday_ = trading_day;
	Manager::Inst().begin_new_tradingday();
	std::this_thread::sleep_for(std::chrono::milliseconds(3000));
	Manager::Inst().init_new_tradingday();
	Manager::Inst().end_new_tradingday();
}

void MDMgr::on_new_tradingday(uint32_t tradingday
	, std::map<const char*, std::shared_ptr<CTPExchangeInfoEx>, XUtil::strless> ctp_exchanges
	, std::map<const char*, std::shared_ptr<CTPProductInfoEx>, XUtil::strless> ctp_products
	, std::map<const char*, std::shared_ptr<CTPCommodityInfoEx>, XUtil::strless> ctp_commoditys
)
{
	init_exchanges_ = std::move(ctp_exchanges);
	init_products_ = std::move(ctp_products);
	init_commoditys_ = std::move(ctp_commoditys);
	on_new_tradingday2(tradingday);
}

void MDMgr::update_category()
{
	std::vector<zqdb::AllProduct> allproduct;
	for(auto& exchange : all_exchange) 
	{
		auto h = ZQDBGetExchange(exchange);
		if (h) {
			allproduct.emplace_back(h);
		}
	}
	//zqdb::AllCode allcode(g_ctp_module);
	zqdb::Category cat;
	cat.RemoveCategory(u8"期货");
	if (cat.OpenOrCreate(u8"期货", ZQDB_HANDLE_TYPE_CATEGORY)) {
#if 1
		{
			std::vector<HZQDB> mian_codes;
			for (auto& produccts : allproduct)
			{
				for (auto hproduct : produccts)
				{
					zqdb::AllCode allcode(hproduct);
					if (!allcode.empty()) {
						mian_codes.push_back(allcode[0]);
					}
				}
			}
			{
				zqdb::Category t(cat.AppendCategory(u8"期货主力", ZQDB_HANDLE_TYPE_CODE));
				for (auto hcode : mian_codes)
				{
					t.AppendValue(hcode);
				}
			}
		}
#else
		{
			std::vector<HZQDB> mian_codes, second_codes;
			for (auto& produccts : allproduct)
			{
				for (auto hproduct : produccts)
				{
					zqdb::AllCode allcode(hproduct);
					if (!allcode.empty()) {
						mian_codes.push_back(allcode[0]);
						//second_codes.push_back(allcode[1]);
					}
				}
			}
			{
				zqdb::Category t(cat.AppendCategory(u8"全部主力", ZQDB_HANDLE_TYPE_CODE));
				for (auto hcode : mian_codes)
				{
					t.AppendValue(hcode);
				}
				for (auto hcode : second_codes)
				{
					t.AppendValue(hcode);
				}
			}
			{
				zqdb::Category t(cat.AppendCategory(u8"主力", ZQDB_HANDLE_TYPE_CODE));
				for (auto hcode : mian_codes)
				{
					t.AppendValue(hcode);
				}
			}
			{
				zqdb::Category t(cat.AppendCategory(u8"次主力", ZQDB_HANDLE_TYPE_CODE));
				for (auto hcode : second_codes)
				{
					t.AppendValue(hcode);
				}
			}
		}
		//
		cat.AppendSeparator();
		//
		for (size_t i = 0, j = allproduct.size(); i < j; i++)
		{
			size_t count = 0;
			for (auto h : allproduct[i])
			{
				zqdb::Product product(h);
				if (product->Type == PRODUCT_TYPE_Futures) {
					count++;
					zqdb::Category t(cat.AppendCategory(product->Name, ZQDB_HANDLE_TYPE_PRODUCT));
					t.AppendValue(product);
				}
			}
			if (count && i < (j - 1)) {
				cat.AppendSeparator();
			}
		}
#endif
	}
}

void MDMgr::on_normal(bool new_trading_day)
{
	if (init_flag_ || normal_flag_) {
		return;
	}
	normal_flag_ = true;

#ifdef _DEBUG
	HZQDB h = ZQDBGetCode("au0", EXCHANGE_SHFE);
	uint32_t date = 0, time = 0;
	time = ZQDBGetOpenTime(h, &date, 3);
	time = ZQDBGetCloseTime(h, &date, 3);
	time = ZQDBGetOpenTime(h, &date, 2);
	time = ZQDBGetCloseTime(h, &date, 2);
	time = ZQDBGetOpenTime(h, &date, 1);
	time = ZQDBGetCloseTime(h, &date, 1);
	time = ZQDBGetOpenTime(h, &date, 0);
	time = ZQDBGetCloseTime(h, &date, 0);
	auto MaxTimePoint = ZQDBGetMaxTimePointCount(h, CYC_1MIN, 0);
	MaxTimePoint = ZQDBGetMaxTimePointCount(h, CYC_5MIN, 0);
	MaxTimePoint = ZQDBGetMaxTimePointCount(h, CYC_DAY, 0);
	size_t now_pos = 0;
	int now_cmp = 0;
	auto pre_tradingday = ZQDBPrevTradeDay(h, tradingday_, 1);//XUtil::PrevDay(tradingday_, 1);
	now_cmp = ZQDBIsNotInTradingTime(h, pre_tradingday, 205959, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, pre_tradingday, 210000, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 23000, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 23001, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 85959, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 90000, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 101500, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 101501, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 102959, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 103000, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 113000, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 113001, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 125959, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 130000, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 132959, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 133000, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 145959, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 150000, &now_pos);
	now_cmp = ZQDBIsNotInTradingTime(h, tradingday_, 150001, &now_pos);
	
#endif//

	update_category();
	if (markets_.empty()) {
		bool online_flag = false;
		if (ZQDBIsTest()) {
			online_flag = true;
		}
		else {
			//使用官方数据
			if (ZQDBIsConnect()) {
				if (ZQDBIsLoginSuccess()) {
					online_flag = true;
				}
			}
		}
		on_online(online_flag);
	}
	else {
		on_online(online_flag_);
		if (ZQDBServer()) {
			register_timer_for_push();
		}
		subscribe();
	}
}

void MDMgr::OnNotifyStatus(HZQDB h)
{
	switch (h->type)
	{
	case ZQDB_HANDLE_TYPE_CLIENT: {
		if (markets_.empty()) {
			//使用官方数据
			uint32_t new_tradingday = 0;
			bool online_flag = false;
			if (ZQDBIsConnect()) {
				ZQDB_MSG* hmsg = nullptr;
				if (ZQDBIsLogin(&hmsg)) {
					zqdb::Msg msg(hmsg, zqdb::Msg::AutoDelete);
					if (!msg.IsError()) {
						//如果模块更新时间变了，这要重新初始化
						auto TradingDay = msg.GetParamAsUInt(STR_MODULE_NAME, 0);
						if (TradingDay != tradingday_) {
							new_tradingday = TradingDay;
						}
						else {
							online_flag = true;
						}
					}
				}
			}
			if (new_tradingday) {
				service().post(boost::bind(&MDMgr::on_new_tradingday2, this, new_tradingday));
			}
			else  if (!normal_flag_ && online_flag) {
				service().post(boost::bind(&MDMgr::on_normal, this, false));
			}
			else {
				service().post(boost::bind(&This::on_online, this, online_flag));
			}
		}
	} break;
	default: {
	} break;
	}
}

void MDMgr::notify(zqdb::Msg& msg)
{
	if (is_stop()) {
		return;
	}
	if (init_flag_) {
		return;
	}
	auto reqtype = msg.GetReqType();
	switch (reqtype)
	{
	case com::zqdb::proto::msg::MSG_NOTIFY_MODULE_STATUS: {
		if (markets_.empty()) {
			service().post(boost::bind(&This::on_module_status, this, msg));
		}
	} break;
	case com::zqdb::proto::msg::MSG_NOTIFY_MD_SUBSCRIBE: {
		service().post(boost::bind(&This::on_update_msg, this, msg));
	} break;
	}
}

void MDMgr::on_module_status(const zqdb::Msg& msg)
{
	com::zqdb::proto::msg::RtnModuleStatus rtn;
	size_t datalen = 0;
	const char* data = msg.GetData(&datalen);
	rtn.ParseFromArray(data, datalen);
	uint32_t tradingday = rtn.status();
	if (tradingday) {
#ifdef _DEBUG
		auto nowdate = XUtil::NowDateTime();
		if (tradingday < nowdate) {
			tradingday = nowdate;
		}
#endif
		service().post(boost::bind(&MDMgr::on_new_tradingday2, this, tradingday));
	}
}

void MDMgr::on_update_msg(const zqdb::Msg& msg)
{
	com::zqdb::proto::msg::RtnSubscribe rtn;
	size_t datalen = 0;
	const char* data = msg.GetData(&datalen);
	rtn.ParseFromArray(data, datalen);
	update_msg(rtn);
#if USE_LOG_FILE
	/*if (ZQDBIsRecording()) {
		uint32_t date = 0, time = 0;
		date = XUtil::NowDateTime(&time);
		tagZQDBRecordHead head;
		head.time = time;
		head.type = RECORD_TYPE_UpdateMarketData;
		const tagZQDBRecordBuf buf[1] = {
			{ data, datalen },
		};
		ZQDBWriteRecordData(g_tora_module, &head, buf, 1);
	}*/
#endif
}

int MDMgr::request(zqdb::Msg& msg)
{
	if (is_stop()) {
		return ZQDB_STATUS_ERROR;
	}
	switch (msg.GetMsgType())
	{
	case ZQDB_MSG_REQUEST_DO: {
		auto req_type = msg.GetReqType();
		switch (req_type)
		{
		case com::zqdb::proto::msg::MSG_REQUEST_DO_MD_SUBSCRIBE: {
			HZQDB peer = msg.GetNode();
			if (peer) {
				//来自网络的请求
				std::set<HZQDB> subs;
				size_t datalen = 0;
				const char* data = msg.GetData(&datalen);
				com::zqdb::proto::msg::ReqSubscribe req;
				req.ParseFromArray(data, datalen);
				if (req.sub_size() == 0 && req.unsub_size() == 0) {
					//全市场订阅
					ZQDBAddSubscribe(peer, &g_ctp_module, 1);
					zqdb::AllCode allcode(g_ctp_module);
					for (auto h : allcode)
					{
						subs.insert(h);
					}
				}
				else {
					//按需订阅
					if (req.sub_size()) {
						for (auto& one : req.sub())
						{
							HZQDB h = ZQDBGetCode(one.code().c_str(), one.exchange().c_str());
							if (h) {
								subs.insert(h);
								ZQDBAddSubscribe(peer, &h, 1);
							}
						}
					}
					if (req.unsub_size()) {
						for (auto& one : req.unsub())
						{
							HZQDB h = ZQDBGetCode(one.code().c_str(), one.exchange().c_str());
							if (h) {
								ZQDBRemoveSubscribe(peer, &h, 1);
							}
						}
					}
				}
				zqdb::Msg rsp(ZQDB_MSG_RESPONSE);
				rsp.SetNode(peer);
				rsp.SetBizType(msg.GetBizType());
				rsp.SetID(msg.GetID());
				rsp.SetError(ZQDB_STATUS_OK);
				ZQDBSendMsg(rsp);
				//第一次推送全部数据
				if (!subs.empty()) {
					com::zqdb::proto::msg::RtnSubscribe rtn;
					MDB_FIELD diff_field[MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE)] = { 0 };
					size_t diff_field_num = zqdb::BuildFullFTDiffField(diff_field);
					diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, YCLOSE);
					diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, YSETTLE);
					diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, UPPER);
					diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, LOWER);
					diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, PreOpenInterest);
					diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, PreDelta);
					diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, CurrDelta);
					for (auto h : subs) {
						zqdb::BuildCodeInfo(rtn.add_codes(), (CTPCODEINFO*)ZQDBGetValue(h), diff_field, diff_field_num);
					}
					if (1) {
						auto strdiff = rtn.SerializeAsString();
						zqdb::Msg notify(ZQDB_MSG_NOTIFY);
						notify.SetReqID(com::zqdb::proto::msg::MSG_NOTIFY_MD_SUBSCRIBE);
						notify.SetNode(peer);
						notify.SetParam("Module", STR_MODULE_NAME);
						//notify.SetParam("Exchange", data->Exchange);
						//notify.SetParam("Code", data->Code);
						notify.SetData(strdiff.data(), strdiff.size());
						ZQDBSendMsg(notify);
					}
				}
			}
			else {
				//本地请求
				if (!normal_flag_) {
					return ZQDB_STATUS_OK;
				}
				if (sub_all_) {
					return ZQDB_STATUS_OK;
				}
				if (markets_.empty()) {
					msg.SetParam("Module", STR_MODULE_NAME);
					ZQDBRequest(msg, nullptr, 0);
				}
				else {
					com::zqdb::proto::msg::ReqSubscribe req;
					size_t datalen = 0;
					const char* data = msg.GetData(&datalen);
					req.ParseFromArray(data, datalen);
					if (req.sub_size() == 0 && req.unsub_size() == 0) {
						//全市场订阅
						ASSERT(0);
					}
					else {
						//按需订阅
						std::shared_ptr<std::set<HZQDB>> subs, unsubs;
						if (req.sub_size()) {
							subs = std::make_shared<std::set<HZQDB>>();
							for (auto& one : req.sub())
							{
								subs->insert(ZQDBGetCode(one.code().c_str(), one.exchange().c_str()));
							}
						}
						if (req.unsub_size()) {
							unsubs = std::make_shared<std::set<HZQDB>>();
							for (auto& one : req.unsub())
							{
								unsubs->insert(ZQDBGetCode(one.code().c_str(), one.exchange().c_str()));
							}
						}
						service().post(boost::bind(&This::subscribe, this, subs, unsubs));
					}
				}
			}
		} break;
		default: {
			
		} break;
		}
	} break;
	case ZQDB_MSG_REQUEST_QUERY: {
		switch (msg.GetReqType())
		{
		case com::zqdb::proto::msg::MSG_REQUEST_QUERY_MD_MARKET: {
			ASSERT(!sub_all_);
			ASSERT(ZQDBIsRPC());
			msg.SetParam("Module", STR_MODULE_NAME);
			ZQDBRequest(msg, nullptr, 0);
			return ZQDB_STATUS_OK;
		} break;
		case com::zqdb::proto::msg::MSG_REQUEST_QUERY_MD_CODE: {
			ASSERT(!sub_all_);
			ASSERT(ZQDBIsRPC());
			msg.SetParam("Module", STR_MODULE_NAME);
			//auto Code = msg.GetParam("Code");
			auto Product = msg.GetParam("Product");
			auto Exchange = msg.GetParam("Exchange");
			if (Product && Product[0]) {
				//查询品种代码
				auto h = ZQDBGetProduct(Product, Exchange);
				if (h) {
					ZQDBUpdateOrInitCodeTable(MDB_FIELD_TABLE(CTP, CODE), MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE), h, MaxCodeCountPerProduct);
					ZQDBRequest(msg, nullptr, 0);
					return ZQDB_STATUS_OK;
				}
			}
			else if (Exchange && Exchange[0]) {
				//查询市场品种
				auto h = ZQDBGetExchange(Exchange);
				if (h) {
					ZQDBUpdateOrInitProductTable(MDB_FIELD_TABLE(ZQDB, PRODUCT), MDB_FIELD_TABLE_FIELD_COUNT(ZQDB, PRODUCT), h, MaxProductCountPerExchange);
					ZQDBRequest(msg, nullptr, 0);
					return ZQDB_STATUS_OK;
				}
			}
			else {
				ZQDBRequest(msg, nullptr, 0);
				return ZQDB_STATUS_OK;
			}
		} break;
		case com::zqdb::proto::msg::MSG_REQUEST_QUERY_MD_DATA: {
			ASSERT(!sub_all_);
			ASSERT(ZQDBClient());
			msg.SetParam("Module", STR_MODULE_NAME);
			/*auto content_type = msg.GetParam("content_type");
			if (!content_type) {
				auto Code = msg.GetParam("Code");
				auto Exchange = msg.GetParam("Exchange");
				auto Period = (PERIODTYPE)msg.GetParamAsInt("Period", CYC_MAX);
				auto h = ZQDBGetCode(Code, Exchange);
				if (h) {
					if (Period == CYC_MAX) {
						//初始化明细和K线数据到DB
						ZQDBUpdateOrInitTickData(h, MDB_FIELD_TABLE(CTP, TICK), MDB_FIELD_TABLE_FIELD_COUNT(CTP, TICK), nullptr, MaxDataReservedCount_[CYC_TICK]);

						for (auto cycle = CYC_MIN; cycle < CYC_MAX; cycle = (PERIODTYPE)(cycle + 1))
						{
							if (!ZQDBIsBaseCycle(cycle)) {
								continue;
							}
							auto MaxTimePoint = ZQDBGetMaxTimePointCount(h, cycle);
							auto max_data_reserved = MaxDataReservedCount_[cycle];
							ZQDBUpdateOrInitKData(h, cycle, MDB_FIELD_TABLE(CTP, KDATA), MDB_FIELD_TABLE_FIELD_COUNT(CTP, KDATA), nullptr, MaxTimePoint + max_data_reserved);
						}
					}
					else if (Period == CYC_TICK) {
						//初始化明细数据到DB
						ZQDBUpdateOrInitTickData(h, MDB_FIELD_TABLE(CTP, TICK), MDB_FIELD_TABLE_FIELD_COUNT(CTP, TICK), nullptr, MaxDataReservedCount_[CYC_TICK]);
					}
					else {
						auto cycle = Period;
						auto MaxTimePoint = ZQDBGetMaxTimePointCount(h, cycle);
						auto max_data_reserved = MaxDataReservedCount_[cycle];
						ZQDBUpdateOrInitKData(h, cycle, MDB_FIELD_TABLE(CTP, KDATA), MDB_FIELD_TABLE_FIELD_COUNT(CTP, KDATA), nullptr, MaxTimePoint + max_data_reserved);
					}
				}
			}*/
			return ZQDBRequest(msg, nullptr, 0);
		} break;
		}
	} break;
	}
	return ZQDB_STATUS_ERROR;
}

void MDMgr::unregister_timer()
{
	boost::system::error_code ec;
	timer_.cancel(ec);
}

void MDMgr::register_timer_for_push(size_t millis)
{
	push_flag_ = false;
	for (auto& pr : pushs_)
	{
		for (size_t i = 0; i < pr.second.size(); i++)
		{
			if (pr.second[i]) {
				pr.second[i] = false;
			}
		}
	}
	timer_.expires_from_now(boost::posix_time::milliseconds(millis));
	timer_.async_wait(boost::bind(&This::on_timer_for_push, this, boost::asio::placeholders::error));
}

void MDMgr::on_timer_for_push(const boost::system::error_code& ec)
{
	if (ec) {
		return;
	}
	if (push_flag_) {
		std::string strdiff;
		bool broadcast = ZQDBIPCIsAnySubscribe(&g_ctp_module, 1) || ZQDBRPCIsAnySubscribe(&g_ctp_module, 1);
		if (broadcast) {
			com::zqdb::proto::msg::RtnSubscribe rtn;
			bool update = false;
			for (const auto& pr : pushs_)
			{
				zqdb::CodeT<CTPCODEINFO> obj(pr.first);
				const auto& code = *(CTPCODEINFO*)obj;
				MDB_FIELD diff_field[MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE)] = { 0 };
				size_t diff_field_num = 0;
				for (size_t i = 0; i < pr.second.size(); i++)
				{
					if (pr.second[i]) {
						update = true;
						diff_field[diff_field_num++] = MDB_FIELD_TABLE(CTP, CODE)[i];
					}
				}
				if (diff_field_num > 0) {
					auto one = rtn.add_codes();
					zqdb::BuildCodeInfo(one, &code, diff_field, diff_field_num);
				}
			}
			if (update) {
				strdiff = rtn.SerializeAsString();
			}
		}
		if (!strdiff.empty()) {
			zqdb::Msg msg(ZQDB_MSG_NOTIFY);
			msg.SetReqID(com::zqdb::proto::msg::MSG_NOTIFY_MD_SUBSCRIBE);
			msg.SetParam("Module", STR_MODULE_NAME);
			//msg.SetParam("Exchange", product->Exchange);
			//msg.SetParam("Product", product->Product);
			//msg.SetParam("Code", code.Code);
			msg.SetData(strdiff.data(), strdiff.size());
			struct BroadcastUserData {
				zqdb::Msg& msg;
				HZQDB h;
			}user_data{ msg, g_ctp_module };
			if (ZQDBIPCIsAnySubscribe(&g_ctp_module, 1)) {
				ZQDBBroadcastIPC([](HZQDB peer, void* user_data) -> bool {
					auto data = (BroadcastUserData*)user_data;
					if (ZQDBPeerIsAnySubscribe(peer, &data->h, 1)) {
						auto& msg = data->msg;
						msg.SetNode(peer);
						ZQDBSendMsg(msg);
					}
					return false;
				}, &user_data);
			}
			if (ZQDBRPCIsAnySubscribe(&g_ctp_module, 1)) {
				ZQDBBroadcastRPC([](HZQDB peer, void* user_data) -> bool {
					auto data = (BroadcastUserData*)user_data;
					if (ZQDBPeerIsAnySubscribe(peer, &data->h, 1)) {
						auto& msg = data->msg;
						msg.SetNode(peer);
						ZQDBSendMsg(msg);
					}
					return false;
				}, &user_data);
			}
		}
	}
	register_timer_for_push();
}

void MDMgr::subscribe()
{
	LOG4I(STR_MODULE_NAME " market subscribe");
	zqdb::AllCode subs(g_ctp_module);
	if (!sub_all_) {
		subs_.clear();
		for (auto h : subs) {
			if (ZQDBIsSubscribeMarketData(h)) {
				subs_.push_back(h);
			}
		}
	}
	else {
		subs_ = subs;
	}
	if (markets_.empty()) {
		//使用官方数据
		if (!sub_all_) {
			if (subs_.empty()) {
				//
			}
			else {
				zqdb::Msg msg(ZQDB_MSG_REQUEST_DO);
				msg.SetReqID(com::zqdb::proto::msg::MSG_REQUEST_DO_MD_SUBSCRIBE);
				msg.SetParam("Module", STR_MODULE_NAME);
				com::zqdb::proto::msg::ReqSubscribe req;
				for (auto one : subs_)
				{
					zqdb::Code code(one);
					auto sub = req.add_sub();
					sub->set_exchange(code->Exchange);
					sub->set_code(code->Code);
				}
				auto data = req.SerializeAsString();
				msg.SetData(data.data(), data.size());
				ZQDBRequest(msg, nullptr, 0);
			}
		}
		else {
			zqdb::Msg msg(ZQDB_MSG_REQUEST_DO);
			msg.SetReqID(com::zqdb::proto::msg::MSG_REQUEST_DO_MD_SUBSCRIBE);
			msg.SetParam("Module", STR_MODULE_NAME);
			com::zqdb::proto::msg::ReqSubscribe req;
			/*zqdb::AllCode allcode(g_ctp_module);
			for (auto one : allcode)
			{
				zqdb::Code code(one);
				auto sub = req.add_sub();
				sub->set_exchange(code->Exchange);
				sub->set_code(code->Code);
			}*/
			auto data = req.SerializeAsString();
			msg.SetData(data.data(), data.size());
			ZQDBRequest(msg, nullptr, 0);
		}
	}
	else {
		for (size_t i = 0; i < markets_.size(); i++)
		{
			markets_[i]->unsubscribe({});
		}
		for (size_t i = 0; i < markets_.size(); i++)
		{
			markets_[i]->subscribe({});
		}
	}
}

void MDMgr::subscribe(std::shared_ptr<std::set<HZQDB>> subs, std::shared_ptr<std::set<HZQDB>> unsubs)
{
	//按需订阅
	std::vector<HZQDB> codes, uncodes;
	if (unsubs) {
		for (auto h : *unsubs)
		{
			auto it = std::find(subs_.begin(), subs_.end(), h);
			if (it != subs_.end()) {
				subs_.erase(it);
				uncodes.push_back(h);
			}
		}
	}
	if (subs) {
		for (auto h : *subs)
		{
			if (std::find(subs_.begin(), subs_.end(), h) == subs_.end()) {
				subs_.push_back(h);
				codes.push_back(h);
			}
		}
	}
	if (markets_.empty()) {
		ASSERT(0);
	}
	else {
		if (!uncodes.empty()) {
			for (auto& market : markets_)
			{
				market->unsubscribe(uncodes);
			}
		}
		if (!codes.empty()) {
			for (auto& market : markets_)
			{
				market->subscribe(codes);
			}
		}
	}
}
//
//int MDMgr::now_data(HZQDB h, PERIODTYPE cycle, void* data)
//{
//	if (cycle == CYC_TICK) {
//		return -1;
//	}
//	zqdb::CodeT<CTPCODEINFO> code(h);
//	auto kdata = (CTPKDATA*)data;
//	ASSERT(tradingday_);
//	kdata->Date = cycle >= CYC_DAY ? tradingday_ : code->Date;
//	kdata->Time = cycle >= CYC_DAY ? 0 : code->Time / 1000;
//	if (IsZeroValue(code->Volume)) {
//		auto price = code->YClose;
//		kdata->Open = price;
//		kdata->High = price;
//		kdata->Low = price;
//		kdata->Close = price;
//	}
//	else {
//		kdata->Open = code->Open;
//		kdata->High = code->High;
//		kdata->Low = code->Low;
//		kdata->Close = code->Close;
//	}
//	kdata->Amount = code->Amount;
//	kdata->Volume = code->Volume;
//	kdata->OpenInterest = code->OpenInterest;
//	return 0;
//}

int MDMgr::init_data(HZQDB h, PERIODTYPE base_cycle, zqdb::Msg& msg, HZQDB base)
{
	if (!h) {
		return -1;
	}
	if (h->type == ZQDB_HANDLE_TYPE_EXCHANGE) {
		if (msg) {
			if (!msg.IsError()) {
				size_t elem_num = msg.GetParamAsUInt("elem_num", 0);
				size_t data_sz = 0;
				auto data = msg.GetData(&data_sz);
				com::mdb::proto::msg::StdTable table;
				if (table.ParseFromArray(data, data_sz) && elem_num) {
					const auto& elem_attr = table.attr();
					const auto& elem_data = table.data();
					const std::string* const* pMsgEXCHANGE = nullptr;
					const std::string* const* pMsgPRODUCT = nullptr;
					const std::string* const* pMsgNAME = nullptr;
					const uint32_t* pMsgTYPE = nullptr;
					const uint32_t* pMsgTRADE_TIME_COUNT = nullptr;
					const uint32_t* pMsgTRADE_TIME_FROM1 = nullptr;
					const uint32_t* pMsgTRADE_TIME_TO1 = nullptr;
					const uint32_t* pMsgTRADE_TIME_FROM2 = nullptr;
					const uint32_t* pMsgTRADE_TIME_TO2 = nullptr;
					const uint32_t* pMsgTRADE_TIME_FROM3 = nullptr;
					const uint32_t* pMsgTRADE_TIME_TO3 = nullptr;
					const uint32_t* pMsgTRADE_TIME_FROM4 = nullptr;
					const uint32_t* pMsgTRADE_TIME_TO4 = nullptr;
					const uint32_t* pMsgSTATUS = nullptr;
					const uint32_t* pMsgENTER_REASON = nullptr;
					const uint32_t* pMsgTRADING_UNIT = nullptr;
					const uint32_t* pMsgVOLUME_MULTIPLE = nullptr;
					for (size_t j = 0; j < elem_attr.size(); j++)
					{
						auto col_id = elem_attr[j].id();
						if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, EXCHANGE)) {
							pMsgEXCHANGE = elem_data[j].str().str().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, PRODUCT)) {
							pMsgPRODUCT = elem_data[j].str().str().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, NAME)) {
							pMsgNAME = elem_data[j].str().str().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, TYPE)) {
							pMsgTYPE = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, TRADE_TIME_COUNT)) {
							pMsgTRADE_TIME_COUNT = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, TRADE_TIME_FROM1)) {
							pMsgTRADE_TIME_FROM1 = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, TRADE_TIME_TO1)) {
							pMsgTRADE_TIME_TO1 = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, TRADE_TIME_FROM2)) {
							pMsgTRADE_TIME_FROM2 = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, TRADE_TIME_TO2)) {
							pMsgTRADE_TIME_TO2 = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, TRADE_TIME_FROM3)) {
							pMsgTRADE_TIME_FROM3 = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, TRADE_TIME_TO3)) {
							pMsgTRADE_TIME_TO3 = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, TRADE_TIME_FROM4)) {
							pMsgTRADE_TIME_FROM4 = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, TRADE_TIME_TO4)) {
							pMsgTRADE_TIME_TO4 = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, STATUS)) {
							pMsgSTATUS = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, ENTER_REASON)) {
							pMsgENTER_REASON = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, TRADING_UNIT)) {
							pMsgTRADING_UNIT = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(ZQDB, PRODUCT, VOLUME_MULTIPLE)) {
							pMsgVOLUME_MULTIPLE = elem_data[j].u32().u32().data();
						}
						else {
							//
						}
					}
					zqdb::Exchange exchange(h);
					for (size_t i = 0; i < elem_num; i++)
					{
						PRODUCTINFO product = { 0 };
						strcpy(product.Exchange, pMsgEXCHANGE ? pMsgEXCHANGE[i]->c_str() : exchange->Exchange);
						if (pMsgPRODUCT) {
							strcpy(product.Product, pMsgPRODUCT[i]->c_str());
						}
						if (pMsgNAME) {
							strcpy(product.Name, pMsgNAME[i]->c_str());
						}
						if (pMsgTYPE) {
							product.Type = pMsgTYPE[i];
						}
						if (pMsgTRADE_TIME_COUNT) {
							product.TradeTimeCount = pMsgTRADE_TIME_COUNT[i];
						}
						if (pMsgTRADE_TIME_FROM1) {
							product.TradeTime[0][0] = pMsgTRADE_TIME_FROM1[i];
						}
						if (pMsgTRADE_TIME_TO1) {
							product.TradeTime[0][1] = pMsgTRADE_TIME_TO1[i];
						}
						if (pMsgTRADE_TIME_FROM2) {
							product.TradeTime[1][0] = pMsgTRADE_TIME_FROM2[i];
						}
						if (pMsgTRADE_TIME_TO2) {
							product.TradeTime[1][1] = pMsgTRADE_TIME_TO2[i];
						}
						if (pMsgTRADE_TIME_FROM3) {
							product.TradeTime[2][0] = pMsgTRADE_TIME_FROM3[i];
						}
						if (pMsgTRADE_TIME_TO3) {
							product.TradeTime[2][1] = pMsgTRADE_TIME_TO3[i];
						}
						if (pMsgTRADE_TIME_FROM4) {
							product.TradeTime[3][0] = pMsgTRADE_TIME_FROM4[i];
						}
						if (pMsgTRADE_TIME_TO4) {
							product.TradeTime[3][1] = pMsgTRADE_TIME_TO4[i];
						}
						if (pMsgSTATUS) {
							product.Status = pMsgSTATUS[i];
						}
						if (pMsgENTER_REASON) {
							product.EnterReason = pMsgENTER_REASON[i];
						}
						if (pMsgTRADING_UNIT) {
							product.TradingUnit = pMsgTRADING_UNIT[i];
						}
						if (pMsgVOLUME_MULTIPLE) {
							product.VolumeMultiple = pMsgVOLUME_MULTIPLE[i];
						}
						auto hproduct = ZQDBUpdateOrInitProduct(h, &product);
						ZQDBUpdateOrInitCodeTable(MDB_FIELD_TABLE(CTP, CODE), MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE), hproduct, MaxCodeCountPerProduct);
					}
				}
			}
		}
	}
	else if (h->type == ZQDB_HANDLE_TYPE_PRODUCT) {
		if (msg) {
			if (!msg.IsError()) {
				size_t elem_num = msg.GetParamAsUInt("elem_num", 0);
				size_t data_sz = 0;
				auto data = msg.GetData(&data_sz);
				com::mdb::proto::msg::StdTable table;
				if (table.ParseFromArray(data, data_sz) && elem_num) {
					const auto& elem_attr = table.attr();
					const auto& elem_data = table.data();
					const std::string* const* pMsgEXCHANGE = nullptr;
					const std::string* const* pMsgPRODUCT = nullptr;
					const std::string* const* pMsgCODE = nullptr;
					const std::string* const* pMsgNAME = nullptr;
					const std::string* const* pMsgTRADE_CODE = nullptr;
					const uint32_t* pMsgDATE = nullptr;
					const uint32_t* pMsgTIME = nullptr;
					const uint32_t* pMsgSTATUS = nullptr;
					const uint32_t* pMsgENTER_REASON = nullptr;
					const double* pMsgPRICETICK = nullptr;
					const double* pMsgYCLOSE = nullptr;
					const double* pMsgYSETTLE = nullptr;
					const double* pMsgSETTLE = nullptr;
					const double* pMsgUPPER = nullptr;
					const double* pMsgLOWER = nullptr;
					const double* pMsgOPEN = nullptr;
					const double* pMsgHIGH = nullptr;
					const double* pMsgLOW = nullptr;
					const double* pMsgCLOSE = nullptr;
					const double* pMsgAMOUNT = nullptr;
					const double* pMsgVOLUME = nullptr;
					const double* pMsgBID1 = nullptr;
					const double* pMsgBID2 = nullptr;
					const double* pMsgBID3 = nullptr;
					const double* pMsgBID4 = nullptr;
					const double* pMsgBID5 = nullptr;
					const double* pMsgBIDV1 = nullptr;
					const double* pMsgBIDV2 = nullptr;
					const double* pMsgBIDV3 = nullptr;
					const double* pMsgBIDV4 = nullptr;
					const double* pMsgBIDV5 = nullptr;
					const double* pMsgASK1 = nullptr;
					const double* pMsgASK2 = nullptr;
					const double* pMsgASK3 = nullptr;
					const double* pMsgASK4 = nullptr;
					const double* pMsgASK5 = nullptr;
					const double* pMsgASKV1 = nullptr;
					const double* pMsgASKV2 = nullptr;
					const double* pMsgASKV3 = nullptr;
					const double* pMsgASKV4 = nullptr;
					const double* pMsgASKV5 = nullptr;
					const double* pMsgPreOpenInterest = nullptr;
					const double* pMsgOpenInterest = nullptr;
					const double* pMsgPreDelta = nullptr;
					const double* pMsgCurrDelta = nullptr;
					const uint32_t* pMsgDeliveryYear = nullptr;
					const uint32_t* pMsgDeliveryMonth = nullptr;
					const double* pMsgMaxMarketOrderVolume = nullptr;
					const double* pMsgMinMarketOrderVolume = nullptr;
					const double* pMsgMaxLimitOrderVolume = nullptr;
					const double* pMsgMinLimitOrderVolume = nullptr;
					const uint32_t* pMsgVolumeMultiple = nullptr;
					const uint32_t* pMsgCreateDate = nullptr;
					const uint32_t* pMsgOpenDate = nullptr;
					const uint32_t* pMsgExpireDate = nullptr;
					const uint32_t* pMsgStartDelivDate = nullptr;
					const uint32_t* pMsgEndDelivDate = nullptr;
					const uint32_t* pMsgInstLifePhase = nullptr;
					const uint32_t* pMsgIsTrading = nullptr;
					const uint32_t* pMsgPositionType = nullptr;
					const uint32_t* pMsgPositionDateType = nullptr;
					const double* pMsgLongMarginRatio = nullptr;
					const double* pMsgShortMarginRatio = nullptr;
					const uint32_t* pMsgMaxMarginSideAlgorithm = nullptr;
					const std::string* const* pMsgUnderlyingInstrID = nullptr;
					const double* pMsgStrikePrice = nullptr;
					const uint32_t* pMsgOptionsType = nullptr;
					const double* pMsgUnderlyingMultiple = nullptr;
					const uint32_t* pMsgCombinationType = nullptr;
					for (size_t j = 0; j < elem_attr.size(); j++)
					{
						auto col_id = elem_attr[j].id();
						if (col_id == MDB_FIELD_INDEX(CTP, CODE, EXCHANGE)) {
							pMsgEXCHANGE = elem_data[j].str().str().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, PRODUCT)) {
							pMsgPRODUCT = elem_data[j].str().str().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, CODE)) {
							pMsgCODE = elem_data[j].str().str().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, NAME)) {
							pMsgNAME = elem_data[j].str().str().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, TRADE_CODE)) {
							pMsgTRADE_CODE = elem_data[j].str().str().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, DATE)) {
							pMsgDATE = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, TIME)) {
							pMsgTIME = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, STATUS)) {
							pMsgSTATUS = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ENTER_REASON)) {
							pMsgENTER_REASON = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, PRICETICK)) {
							pMsgPRICETICK = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, YCLOSE)) {
							pMsgYCLOSE = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, YSETTLE)) {
							pMsgYSETTLE = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, SETTLE)) {
							pMsgSETTLE = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, UPPER)) {
							pMsgUPPER = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, LOWER)) {
							pMsgLOWER = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, OPEN)) {
							pMsgOPEN = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, HIGH)) {
							pMsgHIGH = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, LOW)) {
							pMsgLOW = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, CLOSE)) {
							pMsgCLOSE = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, AMOUNT)) {
							pMsgAMOUNT = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, VOLUME)) {
							pMsgVOLUME = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, BID1)) {
							pMsgBID1 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, BID2)) {
							pMsgBID2 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, BID3)) {
							pMsgBID3 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, BID4)) {
							pMsgBID4 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, BID5)) {
							pMsgBID5 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, BIDV1)) {
							pMsgBIDV1 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, BIDV2)) {
							pMsgBIDV2 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, BIDV3)) {
							pMsgBIDV3 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, BIDV4)) {
							pMsgBIDV4 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, BIDV5)) {
							pMsgBIDV5 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ASK1)) {
							pMsgASK1 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ASK2)) {
							pMsgASK2 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ASK3)) {
							pMsgASK3 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ASK4)) {
							pMsgASK4 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ASK5)) {
							pMsgASK5 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ASKV1)) {
							pMsgASKV1 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ASKV2)) {
							pMsgASKV2 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ASKV3)) {
							pMsgASKV3 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ASKV4)) {
							pMsgASKV4 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ASKV5)) {
							pMsgASKV5 = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, PreOpenInterest)) {
							pMsgPreOpenInterest = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, OpenInterest)) {
							pMsgOpenInterest = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, PreDelta)) {
							pMsgPreDelta = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, CurrDelta)) {
							pMsgCurrDelta = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, DeliveryYear)) {
							pMsgDeliveryYear = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, DeliveryMonth)) {
							pMsgDeliveryMonth = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, MaxMarketOrderVolume)) {
							pMsgMaxMarketOrderVolume = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, MinMarketOrderVolume)) {
							pMsgMinMarketOrderVolume = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, MaxLimitOrderVolume)) {
							pMsgMaxLimitOrderVolume = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, MinLimitOrderVolume)) {
							pMsgMinLimitOrderVolume = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, VolumeMultiple)) {
							pMsgVolumeMultiple = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, CreateDate)) {
							pMsgCreateDate = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, OpenDate)) {
							pMsgOpenDate = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ExpireDate)) {
							pMsgExpireDate = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, StartDelivDate)) {
							pMsgStartDelivDate = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, EndDelivDate)) {
							pMsgEndDelivDate = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, InstLifePhase)) {
							pMsgInstLifePhase = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, IsTrading)) {
							pMsgIsTrading = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, PositionType)) {
							pMsgPositionType = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, PositionDateType)) {
							pMsgPositionDateType = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, LongMarginRatio)) {
							pMsgLongMarginRatio = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, ShortMarginRatio)) {
							pMsgShortMarginRatio = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, MaxMarginSideAlgorithm)) {
							pMsgMaxMarginSideAlgorithm = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, UnderlyingInstrID)) {
							pMsgUnderlyingInstrID = elem_data[j].str().str().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, StrikePrice)) {
							pMsgStrikePrice = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, OptionsType)) {
							pMsgOptionsType = elem_data[j].u32().u32().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, UnderlyingMultiple)) {
							pMsgUnderlyingMultiple = elem_data[j].f64().f64().data();
						}
						else if (col_id == MDB_FIELD_INDEX(CTP, CODE, CombinationType)) {
							pMsgCombinationType = elem_data[j].u32().u32().data();
						}
						else {
							//
						}
					}
					zqdb::Product product(h);
					for (size_t i = 0; i < elem_num; i++)
					{
						CTPCODEINFO code = { 0 };
						strcpy(code.Exchange, pMsgEXCHANGE ? pMsgEXCHANGE[i]->c_str() : product->Exchange);
						strcpy(code.Product, pMsgPRODUCT ? pMsgPRODUCT[i]->c_str() : product->Product);
						if (pMsgCODE) {
							strcpy(code.Code, pMsgCODE[i]->c_str());
						}
						if (pMsgNAME) {
							strcpy(code.Name, pMsgNAME[i]->c_str());
						}
						if (pMsgTRADE_CODE) {
							strcpy(code.TradeCode, pMsgTRADE_CODE[i]->c_str());
						}
						if (pMsgDATE) {
							code.Date = pMsgDATE[i];
						}
						if (pMsgTIME) {
							code.Time = pMsgTIME[i];
						}
						if (pMsgSTATUS) {
							code.Status = pMsgSTATUS[i];
						}
						if (pMsgENTER_REASON) {
							code.EnterReason = pMsgENTER_REASON[i];
						}
						if (pMsgPRICETICK) {
							code.PriceTick = pMsgPRICETICK[i];
						}
						if (pMsgYCLOSE) {
							code.YClose = pMsgYCLOSE[i];
						}
						if (pMsgYSETTLE) {
							code.YSettle = pMsgYSETTLE[i];
						}
						if (pMsgSETTLE) {
							code.Settle = pMsgSETTLE[i];
						}
						if (pMsgUPPER) {
							code.Upper = pMsgUPPER[i];
						}
						if (pMsgLOWER) {
							code.Lower = pMsgLOWER[i];
						}
						if (pMsgOPEN) {
							code.Open = pMsgOPEN[i];
						}
						if (pMsgHIGH) {
							code.High = pMsgHIGH[i];
						}
						if (pMsgLOW) {
							code.Low = pMsgLOW[i];
						}
						if (pMsgCLOSE) {
							code.Close = pMsgCLOSE[i];
						}
						if (pMsgAMOUNT) {
							code.Amount = pMsgAMOUNT[i];
						}
						if (pMsgVOLUME) {
							code.Volume = pMsgVOLUME[i];
						}
						if (pMsgBID1) {
							code.Bid1 = pMsgBID1[i];
						}
						if (pMsgBID2) {
							code.Bid2 = pMsgBID2[i];
						}
						if (pMsgBID3) {
							code.Bid3 = pMsgBID3[i];
						}
						if (pMsgBID4) {
							code.Bid4 = pMsgBID4[i];
						}
						if (pMsgBID5) {
							code.Bid5 = pMsgBID5[i];
						}
						if (pMsgBIDV1) {
							code.BidV1 = pMsgBIDV1[i];
						}
						if (pMsgBIDV2) {
							code.BidV2 = pMsgBIDV2[i];
						}
						if (pMsgBIDV3) {
							code.BidV3 = pMsgBIDV3[i];
						}
						if (pMsgBIDV4) {
							code.BidV4 = pMsgBIDV4[i];
						}
						if (pMsgBIDV5) {
							code.BidV5 = pMsgBIDV5[i];
						}
						if (pMsgASK1) {
							code.Ask1 = pMsgASK1[i];
						}
						if (pMsgASK2) {
							code.Ask2 = pMsgASK2[i];
						}
						if (pMsgASK3) {
							code.Ask3 = pMsgASK3[i];
						}
						if (pMsgASK4) {
							code.Ask4 = pMsgASK4[i];
						}
						if (pMsgASK5) {
							code.Ask5 = pMsgASK5[i];
						}
						if (pMsgASKV1) {
							code.AskV1 = pMsgASKV1[i];
						}
						if (pMsgASKV2) {
							code.AskV2 = pMsgASKV2[i];
						}
						if (pMsgASKV3) {
							code.AskV3 = pMsgASKV3[i];
						}
						if (pMsgASKV4) {
							code.AskV4 = pMsgASKV4[i];
						}
						if (pMsgASKV5) {
							code.AskV5 = pMsgASKV5[i];
						}
						if (pMsgPreOpenInterest) {
							code.PreOpenInterest = pMsgPreOpenInterest[i];
						}
						if (pMsgOpenInterest) {
							code.OpenInterest = pMsgOpenInterest[i];
						}
						if (pMsgPreDelta) {
							code.PreDelta = pMsgPreDelta[i];
						}
						if (pMsgCurrDelta) {
							code.CurrDelta = pMsgCurrDelta[i];
						}
						if (pMsgDeliveryYear) {
							code.DeliveryYear = pMsgDeliveryYear[i];
						}
						if (pMsgDeliveryMonth) {
							code.DeliveryMonth = pMsgDeliveryMonth[i];
						}
						if (pMsgMaxMarketOrderVolume) {
							code.MaxMarketOrderVolume = pMsgMaxMarketOrderVolume[i];
						}
						if (pMsgMinMarketOrderVolume) {
							code.MinMarketOrderVolume = pMsgMinMarketOrderVolume[i];
						}
						if (pMsgMaxLimitOrderVolume) {
							code.MaxLimitOrderVolume = pMsgMaxLimitOrderVolume[i];
						}
						if (pMsgMinLimitOrderVolume) {
							code.MinLimitOrderVolume = pMsgMinLimitOrderVolume[i];
						}
						if (pMsgVolumeMultiple) {
							code.VolumeMultiple = pMsgVolumeMultiple[i];
						}
						if (pMsgCreateDate) {
							code.CreateDate = pMsgCreateDate[i];
						}
						if (pMsgOpenDate) {
							code.OpenDate = pMsgOpenDate[i];
						}
						if (pMsgExpireDate) {
							code.ExpireDate = pMsgExpireDate[i];
						}
						if (pMsgStartDelivDate) {
							code.StartDelivDate = pMsgStartDelivDate[i];
						}
						if (pMsgEndDelivDate) {
							code.EndDelivDate = pMsgEndDelivDate[i];
						}
						if (pMsgInstLifePhase) {
							code.InstLifePhase = pMsgInstLifePhase[i];
						}
						if (pMsgIsTrading) {
							code.IsTrading = pMsgIsTrading[i];
						}
						if (pMsgPositionType) {
							code.PositionType = pMsgPositionType[i];
						}
						if (pMsgPositionDateType) {
							code.PositionDateType = pMsgPositionDateType[i];
						}
						if (pMsgLongMarginRatio) {
							code.LongMarginRatio = pMsgLongMarginRatio[i];
						}
						if (pMsgShortMarginRatio) {
							code.ShortMarginRatio = pMsgShortMarginRatio[i];
						}
						if (pMsgMaxMarginSideAlgorithm) {
							code.MaxMarginSideAlgorithm = pMsgMaxMarginSideAlgorithm[i];
						}
						if (pMsgUnderlyingInstrID) {
							strcpy(code.UnderlyingInstrID, pMsgUnderlyingInstrID[i]->c_str());
						}
						if (pMsgStrikePrice) {
							code.StrikePrice = pMsgStrikePrice[i];
						}
						if (pMsgOptionsType) {
							code.OptionsType = pMsgOptionsType[i];
						}
						if (pMsgUnderlyingMultiple) {
							code.UnderlyingMultiple = pMsgUnderlyingMultiple[i];
						}
						if (pMsgCombinationType) {
							code.CombinationType = pMsgCombinationType[i];
						}
						auto hcode = ZQDBUpdateOrInitCode(h, (CODEINFO*)&code);
					}
				}
			}
		}
	}
	else if (h->type == ZQDB_HANDLE_TYPE_CODE) {
		//bool add_now = false;
		const char* elem_data = nullptr;
		size_t elem_num = msg.GetParamAsUInt("elem_num", 0);
		com::mdb::proto::msg::Table table;
		std::vector<CTPTICK> std_ticks;
		std::vector<CTPKDATA> std_kdatas;
		com::mdb::proto::msg::StdTable std_table;
		if (msg) {
			if (!msg.IsError()) {
				size_t data_sz = 0;
				auto data = msg.GetData(&data_sz);
				if (base_cycle == CYC_TICK) {
					if (std_table.ParseFromArray(data, data_sz)) {
						std_ticks.resize(elem_num);
						const auto& std_elem_attr = std_table.attr();
						const auto& std_elem_data = std_table.data();
						for (size_t i = 0; i < std_ticks.size(); i++)
						{
							for (size_t j = 0; j < std_elem_attr.size(); j++)
							{
								auto col_id = std_elem_attr[j].id();
								if (col_id == MDB_FIELD_INDEX(CTP, TICK, DATE)) {
									std_ticks[i].Date = std_elem_data[j].u32().u32(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, TIME)) {
									std_ticks[i].Time = std_elem_data[j].u32().u32(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, FLAGS)) {
									std_ticks[i].Flags = std_elem_data[j].u32().u32(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, CLOSE)) {
									std_ticks[i].Close = std_elem_data[j].f64().f64(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, AMOUNT)) {
									std_ticks[i].Amount = std_elem_data[j].f64().f64(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, VOLUME)) {
									std_ticks[i].Volume = std_elem_data[j].f64().f64(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, OpenInterest)) {
									std_ticks[i].OpenInterest = std_elem_data[j].f64().f64(i);
								}
								else {
									//
								}
							}
						}
						elem_data = (const char*)std_ticks.data();
					}
					else if (table.ParseFromArray(data, data_sz)) {
						elem_data = table.data().data();
						ASSERT(table.data().size() % sizeof(CTPTICK) == 0);
						elem_num = table.data().size() / sizeof(CTPTICK);
					}
				}
				else {
					if (std_table.ParseFromArray(data, data_sz)) {
						std_kdatas.resize(elem_num);
						const auto& std_elem_attr = std_table.attr();
						const auto& std_elem_data = std_table.data();
						for (size_t i = 0; i < std_kdatas.size(); i++)
						{
							for (size_t j = 0; j < std_elem_attr.size(); j++)
							{
								auto col_id = std_elem_attr.at(j).id();
								if (col_id == MDB_FIELD_INDEX(CTP, KDATA, DATE)) {
									std_kdatas[i].Date = std_elem_data[j].u32().u32(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, TIME)) {
									std_kdatas[i].Time = std_elem_data[j].u32().u32(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, OPEN)) {
									std_kdatas[i].Open = std_elem_data[j].f64().f64(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, HIGH)) {
									std_kdatas[i].High = std_elem_data[j].f64().f64(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, LOW)) {
									std_kdatas[i].Low = std_elem_data[j].f64().f64(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, CLOSE)) {
									std_kdatas[i].Close = std_elem_data[j].f64().f64(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, AMOUNT)) {
									std_kdatas[i].Amount = std_elem_data[j].f64().f64(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, VOLUME)) {
									std_kdatas[i].Volume = std_elem_data[j].f64().f64(i);
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, OpenInterest)) {
									std_kdatas[i].OpenInterest = std_elem_data[j].f64().f64(i);
								}
								else {
									//
								}
							}
						}
						elem_data = (const char*)std_kdatas.data();
					}
					else if (table.ParseFromArray(data, data_sz)) {
						elem_data = table.data().data();
						ASSERT(table.data().size() % sizeof(CTPKDATA) == 0);
						elem_num = table.data().size() / sizeof(CTPKDATA);
					}
					if (elem_num) {
						auto last = (CTPKDATA*)&elem_data[(elem_num - 1) * sizeof(CTPKDATA)];
						size_t LastTimePointCount = 0;
						if (base_cycle < CYC_DAY) {
							//这里的last->Date/last->Time是已经调整过的时间，即往后靠的时间
							uint32_t opendate = 0, opentime = 0;
							opentime = ZQDBGetOpenTime(h, &opendate, 0);
							if (last->Date > opendate 
								|| (last->Date == opendate && last->Time >= opentime)) {
								uint32_t last_date = last->Date, last_time = last->Time;
								XUtil::PrevDateTime(last_date, last_time, ZQDBCycle2Second(base_cycle, 0));
								size_t pos = 0;
								int cmp = ZQDBIsNotInTradingTime(h, last_date, last_time, &pos);
								LastTimePointCount = ZQDBGetTimePointByTime(h, last_time, cmp, pos, base_cycle, 0) + 1;
							}
						}
						else if (last->Date >= tradingday_) {
							LastTimePointCount = 1;
						}
						if (MaxDataReservedCount_[base_cycle]) {
							auto num = std::min<>(MaxDataReservedCount_[base_cycle] + LastTimePointCount, elem_num);
							elem_data = elem_data + (elem_num - num) * sizeof(CTPKDATA);
							elem_num = num;
						}
						else {
							//
						}
						/*if (cycle >= CYC_DAY) {
							zqdb::Code code(h);
							if (last->Date < code->Date) {
								add_now = true;
							}
						}*/
					}
				}
			}
		}
		if (base_cycle == CYC_MAX) {
			ASSERT(0);
		}
		else if (base_cycle == CYC_TICK) {
			//初始化明细数据到DB
			ZQDBUpdateOrInitTickData(h, MDB_FIELD_TABLE(CTP, TICK), MDB_FIELD_TABLE_FIELD_COUNT(CTP, TICK)
				, (TICK*)elem_data, elem_num
				, MaxDataReservedCount_[CYC_TICK]);
		}
		else {
			auto MaxTimePoint = ZQDBGetMaxTimePointCount(h, base_cycle, 0);
			ZQDBUpdateOrInitKData(h, base_cycle, MDB_FIELD_TABLE(CTP, KDATA), MDB_FIELD_TABLE_FIELD_COUNT(CTP, KDATA)
				, (KDATA*)elem_data, elem_num
				, (MaxDataReservedCount_[base_cycle] ? MaxDataReservedCount_[base_cycle] : elem_num)
				+ std::max<>((MaxTimePoint + 10) / 10 * 10, (size_t)100)/*多预留一点空间，防止空间不够导致内存越界*/
			);
			/*if (add_now) {
				CTPKDATA kdata = { 0 };
				now_data(h, (PERIODTYPE)cycle, &kdata);
				ZQDBAddKData(h, (PERIODTYPE)cycle, (KDATA*)&kdata, sizeof(kdata));
			}*/
		}
	}
	else if (h->type == ZQDB_HANDLE_TYPE_DATA) {
		bool init = false;
		HZQDB code = ZQDBDataGetCode(h);
		size_t cycleex = 0;
		PERIODTYPE cycle = ZQDBDataGetCycle(h, &cycleex);
		if (msg) {
			com::mdb::proto::msg::StdTable std_table;
			if (!msg.IsError()) {
				size_t data_sz = 0;
				auto data = msg.GetData(&data_sz);
				if (std_table.ParseFromArray(data, data_sz)) {
					size_t elem_num = msg.GetParamAsUInt("elem_num", 0);
					if (elem_num > 0) {
						init = true;
						const auto& std_elem_attr = std_table.attr();
						const auto& std_elem_data = std_table.data();
						if (cycle == CYC_TICK) {
							const uint32_t* pMsgDate = nullptr;
							const uint32_t* pMsgTime = nullptr;
							const uint32_t* pMsgFlags = nullptr;
							const double* pMsgClose = nullptr;
							const double* pMsgAmount = nullptr;
							const double* pMsgVolume = nullptr;
							const double* pMsgOpenInterest = nullptr;
							for (size_t j = 0; j < std_elem_attr.size(); j++)
							{
								auto col_id = std_elem_attr[j].id();
								if (col_id == MDB_FIELD_INDEX(CTP, TICK, DATE)) {
									pMsgDate = std_elem_data[j].u32().u32().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, TIME)) {
									pMsgTime = std_elem_data[j].u32().u32().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, FLAGS)) {
									pMsgFlags = std_elem_data[j].u32().u32().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, CLOSE)) {
									pMsgClose = std_elem_data[j].f64().f64().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, AMOUNT)) {
									pMsgAmount = std_elem_data[j].f64().f64().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, VOLUME)) {
									pMsgVolume = std_elem_data[j].f64().f64().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, TICK, OpenInterest)) {
									pMsgOpenInterest = std_elem_data[j].f64().f64().data();
								}
								else {
									//
								}
							}
							size_t count = 0, max_count = MaxDataReservedCount_[CYC_TICK];
							ZQDBDataInit(h, MDB_FIELD_TABLE(CTP, TICK)
								, MDB_FIELD_TABLE_FIELD_COUNT(CTP, TICK)
								, max_count);
							uint32_t* pDate = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, DATE));
							uint32_t* pTime = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, TIME));
							uint32_t* pFlags = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, FLAGS));
							double* pClose = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, CLOSE));
							double* pAmount = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, AMOUNT));
							double* pVolume = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, VOLUME));
							double* pOpenInterest = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, OpenInterest));
							size_t pos = elem_num > max_count ? elem_num - max_count : 0;
							count = elem_num > max_count ? max_count : elem_num;
							memcpy(pDate, pMsgDate + pos, count * sizeof(uint32_t));
							memcpy(pTime, pMsgTime + pos, count * sizeof(uint32_t));
							memcpy(pFlags, pMsgFlags + pos, count * sizeof(uint32_t));
							memcpy(pClose, pMsgClose + pos, count * sizeof(double));
							memcpy(pAmount, pMsgAmount + pos, count * sizeof(double));
							memcpy(pVolume, pMsgVolume + pos, count * sizeof(double));
							if(pMsgOpenInterest)
								memcpy(pOpenInterest, pMsgOpenInterest + pos, count * sizeof(double));
							else
								memset(pOpenInterest, 0, count * sizeof(double));
							ZQDBDataSetCount(h, count);
						}
						else {
							const uint32_t* pMsgDate = nullptr;
							const uint32_t* pMsgTime = nullptr;
							const double* pMsgOpen = nullptr;
							const double* pMsgHigh = nullptr;
							const double* pMsgLow = nullptr;
							const double* pMsgClose = nullptr;
							const double* pMsgAmount = nullptr;
							const double* pMsgVolume = nullptr;
							const double* pMsgOpenInterest = nullptr;
							for (size_t j = 0; j < std_elem_attr.size(); j++)
							{
								auto col_id = std_elem_attr[j].id();
								if (col_id == MDB_FIELD_INDEX(CTP, KDATA, DATE)) {
									pMsgDate = std_elem_data[j].u32().u32().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, TIME)) {
									pMsgTime = std_elem_data[j].u32().u32().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, OPEN)) {
									pMsgOpen = std_elem_data[j].f64().f64().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, HIGH)) {
									pMsgHigh = std_elem_data[j].f64().f64().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, LOW)) {
									pMsgLow = std_elem_data[j].f64().f64().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, CLOSE)) {
									pMsgClose = std_elem_data[j].f64().f64().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, AMOUNT)) {
									pMsgAmount = std_elem_data[j].f64().f64().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, VOLUME)) {
									pMsgVolume = std_elem_data[j].f64().f64().data();
								}
								else if (col_id == MDB_FIELD_INDEX(CTP, KDATA, OpenInterest)) {
									pMsgOpenInterest = std_elem_data[j].f64().f64().data();
								}
								else {
									//
								}
							}
							ASSERT(ZQDBIsBaseCycle(cycle) && cycleex == 0);
							auto MaxTimePoint = ZQDBGetMaxTimePointCount(code, cycle, cycleex);
							size_t count = 0, max_count = elem_num + std::max<>((MaxTimePoint + 10) / 10 * 10, (size_t)100); //多预留一点空间，防止空间不够导致内存越界
							ZQDBDataInit(h, MDB_FIELD_TABLE(CTP, KDATA)
								, MDB_FIELD_TABLE_FIELD_COUNT(CTP, KDATA)
								, max_count);
							uint32_t* pDate = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, DATE));
							uint32_t* pTime = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, TIME));
							double* pOpen = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OPEN));
							double* pHigh = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, HIGH));
							double* pLow = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, LOW));
							double* pClose = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, CLOSE));
							double* pAmount = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, AMOUNT));
							double* pVolume = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, VOLUME));
							double* pOpenInterest = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OpenInterest));
							size_t pos = 0;
							count = elem_num;
							memcpy(pDate, pMsgDate + pos, count * sizeof(uint32_t));
							memcpy(pTime, pMsgTime + pos, count * sizeof(uint32_t));
							memcpy(pOpen, pMsgOpen + pos, count * sizeof(double));
							memcpy(pHigh, pMsgHigh + pos, count * sizeof(double));
							memcpy(pLow, pMsgLow + pos, count * sizeof(double));
							memcpy(pClose, pMsgClose + pos, count * sizeof(double));
							memcpy(pAmount, pMsgAmount + pos, count * sizeof(double));
							memcpy(pVolume, pMsgVolume + pos, count * sizeof(double));
							if (pMsgOpenInterest)
								memcpy(pOpenInterest, pMsgOpenInterest + pos, count * sizeof(double));
							else
								memset(pOpenInterest, 0, count * sizeof(double));
							ZQDBDataSetCount(h, count);
						}
					}
				}
			}
		}
		else {
			if (base_cycle == CYC_TICK) {
				ASSERT(ZQDBIsDataExist(code, base_cycle));
			}
			else {
				size_t base_maxcount = 0;
				size_t base_count = 0;
				uint32_t* pBaseDate = nullptr;
				uint32_t* pBaseTime = nullptr;
				double* pBaseOpen = nullptr;
				double* pBaseHigh = nullptr;
				double* pBaseLow = nullptr;
				double* pBaseClose = nullptr;
				double* pBaseAmount = nullptr;
				double* pBaseVolume = nullptr;
				double* pBaseOpenInterest = nullptr;
				if (base) {
					//拼接K线，这里如果需要复权则还要实现复权逻辑
					//走势数据cycleex_ = 1或5表示当日走势或5日走势
					init = true;
					base_maxcount = ZQDBDataGetMaxCount(base);
					base_count = ZQDBDataGetCount(base, nullptr);
					ASSERT(base_maxcount);
					pBaseDate = (uint32_t*)ZQDBDataGetValue(base, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, DATE));
					pBaseTime = (uint32_t*)ZQDBDataGetValue(base, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, TIME));
					pBaseOpen = (double*)ZQDBDataGetValue(base, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OPEN));
					pBaseHigh = (double*)ZQDBDataGetValue(base, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, HIGH));
					pBaseLow = (double*)ZQDBDataGetValue(base, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, LOW));
					pBaseClose = (double*)ZQDBDataGetValue(base, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, CLOSE));
					pBaseAmount = (double*)ZQDBDataGetValue(base, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, AMOUNT));
					pBaseVolume = (double*)ZQDBDataGetValue(base, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, VOLUME));
					pBaseOpenInterest = (double*)ZQDBDataGetValue(base, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OpenInterest));
				}
				else if (ZQDBIsBaseCycle(base_cycle)) {
					init = true;
					base_maxcount = ZQDBGetDataMaxCount(code, base_cycle);
					base_count = ZQDBGetDataCount(code, base_cycle, nullptr);
					ASSERT(base_maxcount);
					pBaseDate = (uint32_t*)ZQDBGetDataValue(code, base_cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, DATE));
					pBaseTime = (uint32_t*)ZQDBGetDataValue(code, base_cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, TIME));
					pBaseOpen = (double*)ZQDBGetDataValue(code, base_cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OPEN));
					pBaseHigh = (double*)ZQDBGetDataValue(code, base_cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, HIGH));
					pBaseLow = (double*)ZQDBGetDataValue(code, base_cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, LOW));
					pBaseClose = (double*)ZQDBGetDataValue(code, base_cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, CLOSE));
					pBaseAmount = (double*)ZQDBGetDataValue(code, base_cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, AMOUNT));
					pBaseVolume = (double*)ZQDBGetDataValue(code, base_cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, VOLUME));
					pBaseOpenInterest = (double*)ZQDBGetDataValue(code, base_cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OpenInterest));
				}
				else {
					ASSERT(0);
					return -1;
				}
				if (init) {
					size_t count = 0, max_count = 0;
					if (cycle == base_cycle) {
						if (cycleex) {
							//走势
							ASSERT(cycle == CYC_1MIN);
							max_count = cycleex * ZQDBGetMaxTimePointCount(code, cycle, cycleex);
						}
						else {
							//复权
							ASSERT(0);
							max_count = base_maxcount;
						}
					}
					else {
						max_count = base_maxcount;
					}
					ZQDBDataInit(h, MDB_FIELD_TABLE(CTP, KDATA)
						, MDB_FIELD_TABLE_FIELD_COUNT(CTP, KDATA)
						, max_count);
					uint32_t* pDate = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, DATE));
					uint32_t* pTime = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, TIME));
					double* pOpen = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OPEN));
					double* pHigh = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, HIGH));
					double* pLow = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, LOW));
					double* pClose = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, CLOSE));
					double* pAmount = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, AMOUNT));
					double* pVolume = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, VOLUME));
					double* pOpenInterest = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OpenInterest));
					if (cycle == base_cycle) {
						if (cycleex) {
							//走势
							auto cross_pos = ZQDBGetTradeCrossDayPos(code);
							auto MaxTimePoint = ZQDBGetMaxTimePointCount(code, cycle, cycleex);
							uint32_t tradeday = 0, date = 0, time = 0;
							time = ZQDBGetOpenTime(code, &date, 0);
							if (cycleex > 1) {
								date = ZQDBPrevTradeDay(code, date, cycleex - 1);
							}
							size_t i = std::lower_bound(pBaseDate, pBaseDate + base_count, date) - pBaseDate;
							if (cross_pos >= 0) {
								//找到date交易日开始的第一个交易时间点
								for (; i < base_count; i++)
								{
									if (pBaseDate[i] != date) {
										break;
									}
									if (pBaseTime[i] >= time) {
										break;
									}
								}
							}
							for (; i < base_count && count < max_count;)
							{
								auto TimePoint = count % MaxTimePoint;
								if (TimePoint == 0) {
									tradeday = ZQDBGetTradeDay(code, pBaseDate[i], pBaseTime[i]);
								}
								pTime[count] = ZQDBGetTimeByTimePoint(code, tradeday, TimePoint, cycle, 0, &pDate[count]);
								if (SameCycle(code, pDate[count], pBaseDate[i]
									, pTime[count], pBaseTime[i], cycle, 0)) {
									pOpen[count] = pBaseOpen[i];
									pHigh[count] = pBaseHigh[i];
									pLow[count] = pBaseLow[i];
									pClose[count] = pBaseClose[i];
									pAmount[count] = pBaseAmount[i];
									pVolume[count] = pBaseVolume[i];
									pOpenInterest[count] = pBaseOpenInterest[i];
									count++;
									i++;
								}
								else {
									if (pDate[count] > pBaseDate[i]
										|| (pDate[count] == pBaseDate[i] && pTime[count] > pBaseTime[i])) {
										//容错一下
										i++;
										continue;
									}
									if (count == 0) {
										pOpen[count] = pBaseOpen[i];
										pHigh[count] = pBaseOpen[i];
										pLow[count] = pBaseOpen[i];
										pClose[count] = pBaseOpen[i];
										pAmount[count] = 0;
										pVolume[count] = 0;
										pOpenInterest[count] = pBaseOpenInterest[i];
									}
									else {
										pOpen[count] = pClose[count - 1];
										pHigh[count] = pClose[count - 1];
										pLow[count] = pClose[count - 1];
										pClose[count] = pClose[count - 1];
										pAmount[count] = 0;
										pVolume[count] = 0;
										pOpenInterest[count] = pOpenInterest[count - 1];
									}
									count++;
								}
							}
						}
						else {
							//复权
							for (size_t i = 0; i < base_count; i++)
							{
								uint32_t nDate = pBaseDate[i];
								uint32_t nTime = pBaseTime[i];
								if (cycle < CYC_DAY) {
									XUtil::PrevDateTime(nDate, nTime, ZQDBCycle2Second(base_cycle, 0));
									nTime = ZQDBGetTimeByTime(code, nTime, cycle, cycleex, &nDate);
								}
								if (i == 0 || !SameCycle(code, pDate[count - 1], nDate, pTime[count - 1], nTime, cycle, cycleex)) {
									pDate[count] = nDate;
									pTime[count] = nTime;
									pOpen[count] = pBaseOpen[i];
									pHigh[count] = pBaseHigh[i];
									pLow[count] = pBaseLow[i];
									pClose[count] = pBaseClose[i];
									pAmount[count] = pBaseAmount[i];
									pVolume[count] = pBaseVolume[i];
									pOpenInterest[count] = pBaseOpenInterest[i];
									count++;
								}
								else {
									pDate[count - 1] = nDate;
									pTime[count - 1] = nTime;
									if (pHigh[count - 1] < pBaseHigh[i])
										pHigh[count - 1] = pBaseHigh[i];
									if (pLow[count - 1] > pBaseLow[i])
										pLow[count - 1] = pBaseLow[i];
									pClose[count - 1] = pBaseClose[i];
									pAmount[count - 1] += pBaseAmount[i];
									pVolume[count - 1] += pBaseVolume[i];
									pOpenInterest[count - 1] += pBaseOpenInterest[i];
								}
							}
						}
					}
					else {
						for (size_t i = 0; i < base_count; i++)
						{
							uint32_t nDate = pBaseDate[i];
							uint32_t nTime = pBaseTime[i];
							if (cycle < CYC_DAY) {
								XUtil::PrevDateTime(nDate, nTime, ZQDBCycle2Second(base_cycle, 0));
								nTime = ZQDBGetTimeByTime(code, nTime, cycle, cycleex, &nDate);
							}
							if (i == 0 || !SameCycle(code, pDate[count - 1], nDate, pTime[count - 1], nTime, cycle, cycleex)) {
								pDate[count] = nDate;
								pTime[count] = nTime;
								pOpen[count] = pBaseOpen[i];
								pHigh[count] = pBaseHigh[i];
								pLow[count] = pBaseLow[i];
								pClose[count] = pBaseClose[i];
								pAmount[count] = pBaseAmount[i];
								pVolume[count] = pBaseVolume[i];
								pOpenInterest[count] = pBaseOpenInterest[i];
								count++;
							}
							else {
								pDate[count - 1] = nDate;
								pTime[count - 1] = nTime;
								if (pHigh[count - 1] < pBaseHigh[i])
									pHigh[count - 1] = pBaseHigh[i];
								if (pLow[count - 1] > pBaseLow[i])
									pLow[count - 1] = pBaseLow[i];
								pClose[count - 1] = pBaseClose[i];
								pAmount[count - 1] += pBaseAmount[i];
								pVolume[count - 1] += pBaseVolume[i];
								pOpenInterest[count - 1] += pBaseOpenInterest[i];
							}
						}
					}
					ZQDBDataSetCount(h, count);
				}
			}
		} 
		if (!init) {
			if (cycle == CYC_TICK) {
				size_t count = 0, max_count = MaxDataReservedCount_[CYC_TICK];
				ZQDBDataInit(h, MDB_FIELD_TABLE(CTP, TICK)
					, MDB_FIELD_TABLE_FIELD_COUNT(CTP, TICK)
					, max_count);
			}
			else {
				ASSERT(ZQDBIsBaseCycle(cycle));
				auto MaxTimePoint = ZQDBGetMaxTimePointCount(code, cycle, cycleex);
				size_t count = 0, max_count = MaxDataReservedCount_[cycle] + std::max<>((MaxTimePoint + 10) / 10 * 10, (size_t)100); //多预留一点空间，防止空间不够导致内存越界
				ZQDBDataInit(h, MDB_FIELD_TABLE(CTP, KDATA)
					, MDB_FIELD_TABLE_FIELD_COUNT(CTP, KDATA)
					, max_count);
				/*uint32_t* pDate = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, DATE));
				uint32_t* pTime = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, TIME));
				double* pOpen = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OPEN));
				double* pHigh = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, HIGH));
				double* pLow = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, LOW));
				double* pClose = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, CLOSE));
				double* pAmount = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, AMOUNT));
				double* pVolume = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, VOLUME));
				double* pOpenInterest = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OpenInterest));
				CTPKDATA kdata = { 0 };
				now_data(code, (PERIODTYPE)cycle, &kdata);
				*pDate = kdata.Date;
				*pTime = kdata.Time;
				*pOpen = kdata.Open;
				*pHigh = kdata.High;
				*pLow = kdata.Low;
				*pClose = kdata.Close;
				*pAmount = kdata.Amount;
				*pVolume = kdata.Volume;
				*pOpenInterest = kdata.OpenInterest;
				ZQDBDataSetCount(h, 1);*/
			}
		}
	}
	else {
	}
	return 0;
}

int MDMgr::update_data(HZQDB h, MDB_FIELD* diff_field, size_t diff_field_num, CODEINFO* data)
{
	return update_data_ex(h, *(CTPCODEINFO*)data, diff_field, diff_field_num);
}

int MDMgr::update_data_ex(HZQDB h, const CTPCODEINFO& new_code, MDB_FIELD* diff_field, size_t diff_field_num)
{
	CTPCODEINFO* old_code = nullptr;
	if (h->type == ZQDB_HANDLE_TYPE_DATA)
		old_code = (CTPCODEINFO*)ZQDBDataGetFieldValue(h, nullptr);
	else
		old_code = (CTPCODEINFO*)ZQDBGetValue(h);
	auto& code = *old_code;
	bool diff = false;
	for (size_t i = 0; i < diff_field_num; i++)
	{
		if (diff_field[i].index != MDB_FIELD_INDEX(ZQDB, CODE, DATE) && diff_field[i].index != MDB_FIELD_INDEX(ZQDB, CODE, TIME)) {
			diff = true;
			break;
		}
	}
	if (diff) {
		bool bVolumeInValid = code.Volume < ZQDB_ZERO_CTL_VALUE || code.Amount < ZQDB_ZERO_CTL_VALUE;
		bool bNewVolumeInValid = new_code.Volume < ZQDB_ZERO_CTL_VALUE || new_code.Amount < ZQDB_ZERO_CTL_VALUE;
		double DiffVolume = 0;
		double DiffAmount = 0;
		if (!bNewVolumeInValid) {
			DiffVolume = new_code.Volume - code.Volume;
			DiffAmount = new_code.Amount - code.Amount;
		}


		if (bNewVolumeInValid) {
			//无量则不更新明细/K线
		}
		else if (DiffVolume < ZQDB_ZERO_CTL_VALUE || DiffAmount < ZQDB_ZERO_CTL_VALUE) {
			//无成交则不更新明细/K线
		}
		else {
			size_t change = 0;
			for (size_t i = 0; i < diff_field_num; i++)
			{
				//const char str[] = "";
				//const size_t id = *(const size_t*)"1232";
				if (diff_field[i].index == MDB_FIELD_INDEX(CTP, CODE, VOLUME)
					|| diff_field[i].index == MDB_FIELD_INDEX(CTP, CODE, AMOUNT)) {
					change |= MARKET_VOLUME_CHANGED;
				}
				else if (diff_field[i].index == MDB_FIELD_INDEX(CTP, CODE, CLOSE)
					|| diff_field[i].index == MDB_FIELD_INDEX(CTP, CODE, OPEN)
					|| diff_field[i].index == MDB_FIELD_INDEX(CTP, CODE, HIGH)
					|| diff_field[i].index == MDB_FIELD_INDEX(CTP, CODE, LOW)) {
					change |= MARKET_PRICE_CHANGED;
				}
				else {
					change |= MARKET_OTHER_CHANGED;
				}
			}
			if (change & MARKET_TICK_CHANGED) {
				if (h->type == ZQDB_HANDLE_TYPE_DATA) {
					size_t cycleex = 0;
					PERIODTYPE cycle = ZQDBDataGetCycle(h, &cycleex);
					size_t elem_sz = 1;//获取真实大小
					size_t count = ZQDBDataGetCount(h, &elem_sz);
					if (cycle == CYC_TICK) {
						double close = new_code.Close;
						if (IsInvalidValue(close)) {
							/*if (!IsInvalidValue(new_code.Settle)) {
							close = new_code.Settle;
							}
							else */if (!IsInvalidValue(new_code.Open)) {
								close = new_code.Open;
							}
							else if (!IsInvalidValue(new_code.YSettle)) {
								close = new_code.YSettle;
							}
							else if (!IsInvalidValue(new_code.YClose)) {
								close = new_code.YClose;
							}
							ASSERT(!IsInvalidValue(close));
						}
						uint32_t* pDate = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, DATE));
						uint32_t* pTime = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, TIME));
						uint32_t* pFlags = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, FLAGS));
						double* pClose = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, CLOSE));
						double* pAmount = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, AMOUNT));
						double* pVolume = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, VOLUME));
						double* pOpenInterest = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, TICK) + MDB_FIELD_INDEX(CTP, TICK, OpenInterest));
						if (count >= MaxDataReservedCount_[CYC_TICK]) {
							ZQDBDataRemoveValue(h, 1);
						}
						else {
							count += 1;
						}
						pDate[count - 1] = new_code.Date;
						pTime[count - 1] = new_code.Time;
						pFlags[count - 1] = ZQDBIsBoughtSold(new_code.Bid1, new_code.Ask1, close);
						pClose[count - 1] = close;
						pVolume[count - 1] = DiffVolume;
						pAmount[count - 1] = DiffAmount;
						pOpenInterest[count - 1] = new_code.OpenInterest;
						ZQDBDataSetCount(h, count);
					}
				}
				else {
					{
						double close = new_code.Close;
						if (IsInvalidValue(close)) {
							/*if (!IsInvalidValue(new_code.Settle)) {
							close = new_code.Settle;
							}
							else */if (!IsInvalidValue(new_code.Open)) {
								close = new_code.Open;
							}
							else if (!IsInvalidValue(new_code.YSettle)) {
								close = new_code.YSettle;
							}
							else if (!IsInvalidValue(new_code.YClose)) {
								close = new_code.YClose;
							}
							ASSERT(!IsInvalidValue(close));
						}
						CTPTICK tick;
						tick.Date = new_code.Date;
						tick.Time = new_code.Time;
						tick.Flags = ZQDBIsBoughtSold(new_code.Bid1, new_code.Ask1, close);
						tick.Close = close;
						tick.Amount = DiffAmount;
						tick.Volume = DiffVolume;
						tick.OpenInterest = new_code.OpenInterest;
						ZQDBAddTickData(h, (TICK*)&tick, sizeof(tick));
					}
				}
			}
			if (change & MARKET_TREND_CHANGED) {
				if (h->type == ZQDB_HANDLE_TYPE_DATA) {
					zqdb::Code code(ZQDBDataGetCode(h));
					zqdb::Product product(code.GetProduct()); 
					size_t cycleex = 0;
					PERIODTYPE cycle = ZQDBDataGetCycle(h, &cycleex);
					size_t elem_sz = 1;//获取真实大小
					size_t count = ZQDBDataGetCount(h, &elem_sz);
					if (cycle != CYC_TICK) {
						ASSERT(tradingday_);
						auto Date = cycle >= CYC_DAY ? tradingday_ : new_code.Date;
						auto Time = cycle >= CYC_DAY ? 0 : new_code.Time / 1000;
						auto nDate = Date;
						auto nTime = cycle >= CYC_DAY ? 0 : ZQDBGetTimeByTime(code, Time, cycle, cycleex, &nDate);
						uint32_t* pDate = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, DATE));
						uint32_t* pTime = (uint32_t*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, TIME));
						double* pOpen = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OPEN));
						double* pHigh = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, HIGH));
						double* pLow = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, LOW));
						double* pClose = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, CLOSE));
						double* pAmount = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, AMOUNT));
						double* pVolume = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, VOLUME));
						double* pOpenInterest = (double*)ZQDBDataGetValue(h, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OpenInterest));
						//
						bool bUpdate = true;
						bool bKDataInValid = false;
						if (count == 0) {
							//
						}
						else {
							if (IsInvalidValue(pOpen[count - 1])
								|| IsInvalidValue(pHigh[count - 1])
								|| IsInvalidValue(pLow[count - 1])
								|| IsInvalidValue(pClose[count - 1])) {
								bKDataInValid = true;
							}
							else {
								if (nDate < pDate[count - 1] ||
									(nDate == pDate[count - 1] && nTime < pTime[count - 1] && cycle < CYC_DAY)) {
									bUpdate = false;
								}
								//ASSERT(!(IsZeroValue(kdata.Open) || IsZeroValue(kdata.High) || IsZeroValue(kdata.Low) || IsZeroValue(kdata.Close)));
							}
						}
						if (bUpdate) {
							if (count == 0
								|| (!bKDataInValid
								&& (cycle >= CYC_DAY || (!bVolumeInValid && !bNewVolumeInValid))
								&& !SameCycle(code, pDate[count - 1]
									, nDate, pTime[count - 1], nTime, (PERIODTYPE)cycle, 0)
									)
								) {
								if (cycle == CYC_1MIN && cycleex) {
									//走势
									pTime[count] = ZQDBGetTimeByTimePoint(code, tradingday_, count, cycle, 0, &pDate[count]);
									//先补齐
									if (pDate[count] < nDate
										|| (pDate[count] == nDate && pTime[count] < nTime)) {
										do {
											if (count == 0) {
												pOpen[count] = new_code.Open;
												pHigh[count] = new_code.Open;
												pLow[count] = new_code.Open;
												pClose[count] = new_code.Open;
												pAmount[count] = 0;
												pVolume[count] = 0;
												pOpenInterest[count] = new_code.PreOpenInterest;
											}
											else {
												pOpen[count] = pClose[count - 1];
												pHigh[count] = pClose[count - 1];
												pLow[count] = pClose[count - 1];
												pClose[count] = pClose[count - 1];
												pAmount[count] = 0;
												pVolume[count] = 0;
												pOpenInterest[count] = pOpenInterest[count - 1];
											}
											count++;
											pTime[count] = ZQDBGetTimeByTimePoint(code, tradingday_, count, cycle, 0, &pDate[count]);
										} while (pDate[count] < nDate
											|| (pDate[count] == nDate && pTime[count] < nTime));
									}
									//再新增
									if (count == 0) {
										pOpen[count] = new_code.Open;
										pHigh[count] = new_code.High;
										pLow[count] = new_code.Low;
										pClose[count] = new_code.Close;
										pVolume[count] = new_code.Volume;
										pAmount[count] = new_code.Amount;
									}
									else {
										pOpen[count] = new_code.Close;
										pHigh[count] = new_code.Close;
										pLow[count] = new_code.Close;
										pClose[count] = new_code.Close;
										pVolume[count] = DiffVolume;
										pAmount[count] = DiffAmount;
									}
									pOpenInterest[count] = new_code.PreOpenInterest;
									count++;
									ZQDBDataSetCount(h, count);
								}
								else {
									pDate[count] = nDate;
									pTime[count] = nTime;
									if (count == 0 || cycle >= CYC_DAY) {
										pOpen[count] = new_code.Open;
										pHigh[count] = new_code.High;
										pLow[count] = new_code.Low;
										pClose[count] = new_code.Close;
										pVolume[count] = new_code.Volume;
										pAmount[count] = new_code.Amount;
									}
									else {
										pOpen[count] = new_code.Close;
										pHigh[count] = new_code.Close;
										pLow[count] = new_code.Close;
										pClose[count] = new_code.Close;
										pVolume[count] = DiffVolume;
										pAmount[count] = DiffAmount;
									}
									pOpenInterest[count] = new_code.OpenInterest;
									count++;
									ZQDBDataSetCount(h, count);
								}
							}
							else {
								if (cycle == CYC_DAY) {
									pOpen[count - 1] = new_code.Open;
									pHigh[count - 1] = new_code.High;
									pLow[count - 1] = new_code.Low;
									pClose[count - 1] = new_code.Close;
									pVolume[count - 1] = new_code.Volume;
									pAmount[count - 1] = new_code.Amount;
								}
								else {
									if (bVolumeInValid || bNewVolumeInValid || bKDataInValid) {
										pOpen[count - 1] = new_code.Open;
										pHigh[count - 1] = new_code.High;
										pLow[count - 1] = new_code.Low;
									}
									else {
										if (pHigh[count - 1] < new_code.Close)
											pHigh[count - 1] = new_code.Close;
										if (pLow[count - 1] > new_code.Close)
											pLow[count - 1] = new_code.Close;
									}
									pClose[count - 1] = new_code.Close;
									pVolume[count - 1] += DiffVolume;
									pAmount[count - 1] += DiffAmount;
								}
								pOpenInterest[count - 1] = new_code.OpenInterest;
							}
						}
					}
				}
				else {
					for (size_t cycle = 0; cycle < CYC_MAX; cycle++)
					{
						if (!ZQDBIsBaseCycle((PERIODTYPE)cycle)) {
							continue;
						}
						ASSERT(tradingday_);
						auto Date = cycle >= CYC_DAY ? tradingday_ : new_code.Date;
						auto Time = cycle >= CYC_DAY ? 0 : new_code.Time / 1000;
						auto nDate = Date;
						auto nTime = cycle >= CYC_DAY ? 0 : ZQDBGetTimeByTime(h, Time, (PERIODTYPE)cycle, 0, &nDate);
						bool bUpdate = true;
						bool bKDataInValid = false;
						CTPKDATA kdata = { 0 };
						size_t elem_sz = 1;//获取真实大小
						auto kcount = ZQDBGetDataCount(h, (PERIODTYPE)cycle, &elem_sz);
						if (kcount <= 0) {
							//continue;
						}
						else {
							auto pDate = (uint32_t*)ZQDBGetDataValue(h, (PERIODTYPE)cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, DATE));
							auto pTime = (uint32_t*)ZQDBGetDataValue(h, (PERIODTYPE)cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, TIME));
							auto pOpen = (double*)ZQDBGetDataValue(h, (PERIODTYPE)cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OPEN));
							auto pHigh = (double*)ZQDBGetDataValue(h, (PERIODTYPE)cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, HIGH));
							auto pLow = (double*)ZQDBGetDataValue(h, (PERIODTYPE)cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, LOW));
							auto pClose = (double*)ZQDBGetDataValue(h, (PERIODTYPE)cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, CLOSE));
							auto pAmount = (double*)ZQDBGetDataValue(h, (PERIODTYPE)cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, AMOUNT));
							auto pVolume = (double*)ZQDBGetDataValue(h, (PERIODTYPE)cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, VOLUME));
							auto pOpenInterest = (double*)ZQDBGetDataValue(h, (PERIODTYPE)cycle, MDB_FIELD_TABLE(CTP, KDATA) + MDB_FIELD_INDEX(CTP, KDATA, OpenInterest));
							kdata.Date = pDate[kcount - 1];
							kdata.Time = pTime[kcount - 1];
							kdata.Open = pOpen[kcount - 1];
							kdata.High = pHigh[kcount - 1];
							kdata.Low = pLow[kcount - 1];
							kdata.Close = pClose[kcount - 1];
							kdata.Amount = pAmount[kcount - 1];
							kdata.Volume = pVolume[kcount - 1];
							kdata.OpenInterest = pOpenInterest[kcount - 1];
							if (kdata.Date == 0 || IsInvalidValue(kdata.Open)
								|| IsInvalidValue(kdata.High)
								|| IsInvalidValue(kdata.Low)
								|| IsInvalidValue(kdata.Close)) {
								bKDataInValid = true;
							}
							else {
								if (nDate < kdata.Date ||
									(nDate == kdata.Date && nTime < kdata.Time && cycle < CYC_DAY)) {
									bUpdate = false;
								}
								//ASSERT(!(IsZeroValue(kdata.Open) || IsZeroValue(kdata.High) || IsZeroValue(kdata.Low) || IsZeroValue(kdata.Close)));
							}
						}
						if (bUpdate) {
							if (kcount == 0
								|| (!bKDataInValid
									&& (cycle >= CYC_DAY || (!bVolumeInValid && !bNewVolumeInValid))
									&& !SameCycle(h, kdata.Date, nDate
										, kdata.Time, nTime, (PERIODTYPE)cycle, 0)
									)
								) {
								kdata.Date = nDate;
								kdata.Time = nTime;
								if (kcount == 0 || cycle >= CYC_DAY) {
									kdata.Open = new_code.Open;
									kdata.High = new_code.High;
									kdata.Low = new_code.Low;
									kdata.Close = new_code.Close;
									kdata.Volume = new_code.Volume;
									kdata.Amount = new_code.Amount;
									kdata.OpenInterest = new_code.OpenInterest;
								}
								else {
									kdata.Open = new_code.Close;
									kdata.High = new_code.Close;
									kdata.Low = new_code.Close;
									kdata.Close = new_code.Close;
									kdata.Volume = DiffVolume;
									kdata.Amount = DiffAmount;
									kdata.OpenInterest = new_code.OpenInterest;
								}
								ZQDBAddKData(h, (PERIODTYPE)cycle, (KDATA*)&kdata, sizeof(kdata));
							}
							else {
								if (cycle == CYC_DAY) {
									kdata.Open = new_code.Open;
									kdata.High = new_code.High;
									kdata.Low = new_code.Low;
									kdata.Close = new_code.Close;
									kdata.Volume = new_code.Volume;
									kdata.Amount = new_code.Amount;
								}
								else {
									if (bVolumeInValid || bNewVolumeInValid || bKDataInValid) {
										kdata.Open = new_code.Open;
										kdata.High = new_code.High;
										kdata.Low = new_code.Low;
									}
									else {
										if (kdata.High < new_code.Close)
											kdata.High = new_code.Close;
										if (kdata.Low > new_code.Close)
											kdata.Low = new_code.Close;
									}
									kdata.Close = new_code.Close;
									kdata.Volume += DiffVolume;
									kdata.Amount += DiffAmount;
								}
								kdata.OpenInterest = new_code.OpenInterest;
								ZQDBUpdateKData(h, (PERIODTYPE)cycle, (KDATA*)&kdata, sizeof(kdata));
							}
						}
					}
				}
			}
		}
	}
	return 0;
}

int MDMgr::build_data(HZQDB h, PERIODTYPE cycle, zqdb::Msg& msg)
{
	if (!h) {
		return -1;
	}
	if (h->type == ZQDB_HANDLE_TYPE_EXCHANGE) {
		zqdb::AllProduct allproduct(h);
		std::array<MDB_FIELD, 5> fields = {
			MDB_FIELD_TABLE_FIELD(ZQDB, PRODUCT, PRODUCT),
			MDB_FIELD_TABLE_FIELD(ZQDB, PRODUCT, NAME),
			MDB_FIELD_TABLE_FIELD(ZQDB, PRODUCT, TYPE),
			//MDB_FIELD_TABLE_FIELD(ZQDB, PRODUCT, STATUS),
			//MDB_FIELD_TABLE_FIELD(ZQDB, PRODUCT, ENTER_REASON),
			MDB_FIELD_TABLE_FIELD(ZQDB, PRODUCT, TRADING_UNIT),
			MDB_FIELD_TABLE_FIELD(ZQDB, PRODUCT, VOLUME_MULTIPLE),
		};
		zqdb::BuildStdTableMsg(msg
			, fields.data()
			, fields.size()
			//, MDB_FIELD_TABLE(ZQDB, PRODUCT)
			//, MDB_FIELD_TABLE_FIELD_COUNT(ZQDB, PRODUCT)
			, allproduct.data(), allproduct.size());
	}
	else if (h->type == ZQDB_HANDLE_TYPE_PRODUCT) {
		zqdb::AllCode allcode(h);
		//这里初始化静态信息，行情数据等行情推送更新
		std::array<MDB_FIELD, 30> fields = {
			MDB_FIELD_TABLE_FIELD(CTP, CODE, CODE),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, NAME),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, TRADE_CODE),
			//MDB_FIELD_TABLE_FIELD(CTP, CODE, DATE),
			//MDB_FIELD_TABLE_FIELD(CTP, CODE, TIME),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, PRICETICK),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, YCLOSE),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, YSETTLE),
			//MDB_FIELD_TABLE_FIELD(CTP, CODE, UPPER),
			//MDB_FIELD_TABLE_FIELD(CTP, CODE, LOWER),
			//MDB_FIELD_TABLE_FIELD(CTP, CODE, PreOpenInterest),
			//MDB_FIELD_TABLE_FIELD(CTP, CODE, PreDelta),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, DeliveryYear),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, DeliveryMonth),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, MaxMarketOrderVolume),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, MinMarketOrderVolume),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, MaxLimitOrderVolume),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, MinLimitOrderVolume),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, VolumeMultiple),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, CreateDate),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, OpenDate),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, ExpireDate),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, StartDelivDate),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, EndDelivDate),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, InstLifePhase),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, IsTrading),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, PositionType),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, PositionDateType),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, LongMarginRatio),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, ShortMarginRatio),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, MaxMarginSideAlgorithm),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, UnderlyingInstrID),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, StrikePrice),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, OptionsType),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, UnderlyingMultiple),
			MDB_FIELD_TABLE_FIELD(CTP, CODE, CombinationType),
		};
		zqdb::BuildStdTableMsg(msg
			, fields.data()
			, fields.size()
			//, MDB_FIELD_TABLE(CTP, CODE)
			//, MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE)
			, allcode.data(), allcode.size());
	}
	else if (h->type == ZQDB_HANDLE_TYPE_CODE) {
		if (cycle == CYC_TICK) {
			zqdb::BuildDataMsg(msg, h, cycle
				, MDB_FIELD_TABLE(CTP, TICK)
				, MDB_FIELD_TABLE_FIELD_COUNT(CTP, TICK));
		}
		else {
			zqdb::BuildDataMsg(msg, h, cycle
				, MDB_FIELD_TABLE(CTP, KDATA)
				, MDB_FIELD_TABLE_FIELD_COUNT(CTP, KDATA));
		}
	}
	return 0;
}

bool MDMgr::exist_data(HZQDB h, PERIODTYPE cycle)
{
	if (cycle == CYC_TICK) {
		return false;
	}
	auto dir = ZQDBGetDataDir();
	if (!dir || !dir[0]) {
		return false;
	}
	try {
		zqdb::Code code(h);
		char path[1024] = { 0 };
		snprintf(path, 1024, "%s/%s/market/%s.%s.%s.dat", dir, STR_MODULE_NAME, code->Code, code->Exchange, ZQDBCycle2Str(cycle));
		boost::system::error_code ec;
		auto tt = boost::filesystem::last_write_time(path, ec);
		if (!ec) {
			uint32_t pre_tradeday = ZQDBPrevTradeDay(code, tradingday_, 1);
			uint32_t date = 0;
			uint32_t time = XUtil::MakeTime(tt, &date);
			uint32_t open_date = 0;
			uint32_t open_time = ZQDBGetOpenTime(code, &open_date, (size_t)0);
			uint32_t close_date = 0;
			uint32_t close_time = ZQDBGetCloseTime(code, &close_date, (size_t)-1);
			if ((code->Date == open_date && code->Time >= open_time * 1000)
				|| (code->Date > open_date && code->Date < close_date)
				|| (code->Date == close_date && code->Time <= close_time * 1000)) {
				//交易日时段内
				if (cycle == CYC_DAY) {
					//前一交易日收盘后都算有效文件
					if ((date == pre_tradeday && time > close_time)
						|| (date > pre_tradeday)
						) {
						return true;
					}
				}
			}
			else {
				//交易日时段外
				if (cycle == CYC_DAY) {
					//前一交易日收盘后都算有效文件
					if ((date == pre_tradeday && time > close_time)
						|| (date > pre_tradeday)
						) {
						return true;
					}
				}
				else {
					//前一交易日收盘后到今日开盘前都算有效文件
					//当前交易日收盘后都算有效文件
					if ((date == pre_tradeday && time > close_time)
						|| (date > pre_tradeday && date < open_date)
						|| (date == open_date && time < open_time)
						|| (date == close_date && time > close_time)
						|| (date > close_date)) {
						return true;
					}
				}
			}
			//无效文件则删除
			boost::filesystem::remove(path, ec);
		}
	}
	catch (...) {
	}
	return false;
}

int MDMgr::load_data(HZQDB h, PERIODTYPE cycle, std::string& data)
{
	auto dir = ZQDBGetDataDir();
	if (!dir || !dir[0]) {
		return -1;
	}
	char buf[1024] = { 0 };
	if (h->type == ZQDB_HANDLE_TYPE_EXCHANGE) {
		zqdb::Exchange exchange(h);
		snprintf(buf, 1024, "%s/%s/market/%d/%s.dat", dir, STR_MODULE_NAME
			, tradingday_, exchange->Exchange);
	}
	else if (h->type == ZQDB_HANDLE_TYPE_PRODUCT) {
		zqdb::Product product(h);
		snprintf(buf, 1024, "%s/%s/market/%d/%s.%s.dat", dir, STR_MODULE_NAME
			, tradingday_, product->Product, product->Exchange);
	}
	else if (h->type == ZQDB_HANDLE_TYPE_CODE) {
		if (cycle == CYC_TICK) {
			return -2;
		}
		zqdb::Code code(h);
		snprintf(buf, 1024, "%s/%s/market/%s.%s.%s.dat", dir, STR_MODULE_NAME, code->Code, code->Exchange, ZQDBCycle2Str(cycle));
	}
	//
	try {
		std::ifstream fp(buf, std::ifstream::binary);
		if (fp.is_open()) {
			size_t size;
			fp.seekg(0, std::ios::end);
			size = fp.tellg();
			fp.seekg(0, std::ios::beg);
			data.resize(size);
			fp.read((char*)data.data(), data.size());
			fp.close();
			return 0;
		}
	}
	catch (...) {
	}
	return -3;
}

int MDMgr::save_data(HZQDB h, PERIODTYPE cycle, zqdb::Msg& msg)
{
	auto dir = ZQDBGetDataDir();
	if (!dir || !dir[0]) {
		return -1;
	}
	char buf[1024] = { 0 };
	if (h->type == ZQDB_HANDLE_TYPE_EXCHANGE) {
		zqdb::Exchange exchange(h);
		snprintf(buf, 1024, "%s/%s/market/%d/%s.dat", dir, STR_MODULE_NAME
			, tradingday_, exchange->Exchange);
	}
	else if (h->type == ZQDB_HANDLE_TYPE_PRODUCT) {
		zqdb::Product product(h);
		snprintf(buf, 1024, "%s/%s/market/%d/%s.%s.dat", dir, STR_MODULE_NAME
			, tradingday_, product->Product, product->Exchange);
	}
	else if (h->type == ZQDB_HANDLE_TYPE_CODE) {
		if (cycle == CYC_TICK) {
			return -2;
		}
		zqdb::Code code(h);
		snprintf(buf, 1024, "%s/%s/market/%s.%s.%s.dat", dir, STR_MODULE_NAME, code->Code, code->Exchange, ZQDBCycle2Str(cycle));
	}
	try {
		size_t data_sz = 0;
		const char* data = msg.Serialize(&data_sz);
		if (data && data_sz) {
			std::ofstream ofs(buf, std::ofstream::binary | std::ofstream::trunc);
			ofs.write(data, data_sz);
		}
		return 0;
	}
	catch (...) {

	}
	return -3;
}

int MDMgr::close_data(HZQDB h, PERIODTYPE cycle)
{
	if (h->type == ZQDB_HANDLE_TYPE_DATA) {
		//ASSERT(ZQDBDataIsFieldValue(h));
		size_t cycleex = 0;
		PERIODTYPE cycle = ZQDBDataGetCycle(h, &cycleex);
		if (cycle == CYC_TICK) {
			//清理所有明细
			ZQDBDataSetCount(h, 0);
		}
		else {
			HZQDB code = ZQDBDataGetCode(h);
			if (cycle == CYC_1MIN && cycleex) {
				//走势，清理一个交易日空间
				auto MaxTimePoint = ZQDBGetMaxTimePointCount(code, cycle, cycleex);
				auto MaxValueCount = ZQDBDataGetMaxCount(h);
				ASSERT(MaxTimePoint <= MaxValueCount);
				size_t elem_sz = 1;//获取真实大小
				auto CurValueCount = ZQDBDataGetCount(h, &elem_sz);
				ZQDBDataRemoveValue(h, CurValueCount >= MaxTimePoint ? MaxTimePoint : CurValueCount);
			}
			else {
				//size_t mutiple = 0;
				//auto base_cycle = ZQDBCalcBaseCycle(cycle, cycleex, &mutiple);
				//清理出至少一个交易日的空间
				auto MaxTimePoint = ZQDBGetMaxTimePointCount(code, cycle, cycleex);
				MaxTimePoint = (MaxTimePoint + 10) / 10 * 10; //多预留一点空间，防止空间不够导致内存越界
				auto MaxValueCount = ZQDBDataGetMaxCount(h);
				ASSERT(MaxTimePoint <= MaxValueCount);
				size_t elem_sz = 1;//获取真实大小
				auto CurValueCount = ZQDBDataGetCount(h, &elem_sz);
				if ((CurValueCount + MaxTimePoint) > MaxValueCount) {
					if (MaxDataReservedCount_[cycle]) {
						ZQDBDataRemoveValue(h,
							std::max<>(MaxTimePoint - (MaxValueCount - CurValueCount), (size_t)100) //至少移除100条K线，这样可以避免日线这种大周期K线每天清理
						);
					}
					else {
						ZQDBDataReserveCount(h
							, MaxValueCount + std::max<>(MaxTimePoint, (size_t)100) //至少预留100条K线，这样可以避免日线这种大周期K线每天清理
						);
					}
				}
			}
		}
	}
	else {
		size_t CurValueCount = 0;
		{
			//zqdb::DataWriteGuard lock(h, cycle);
			size_t elem_sz = 1;//获取真实大小
			CurValueCount = ZQDBGetDataCount(h, cycle, &elem_sz);
			if (cycle == CYC_TICK) {
				//清理所有明细
				ZQDBRemoveData(h, cycle, 0, CurValueCount);
			}
			else {
				//清理出至少一个交易日的空间
				auto MaxTimePoint = ZQDBGetMaxTimePointCount(h, cycle, 0);
				MaxTimePoint = (MaxTimePoint + 10) / 10 * 10;; //多预留一点空间，防止空间不够导致内存越界
				auto MaxValueCount = ZQDBGetDataMaxCount(h, cycle);
				ASSERT(MaxTimePoint <= MaxValueCount);
				if ((CurValueCount + MaxTimePoint) > MaxValueCount) {
					if (MaxDataReservedCount_[cycle]) {
						ZQDBRemoveData(h, cycle, 0
							, std::max<>(MaxTimePoint - (MaxValueCount - CurValueCount), (size_t)100) //至少移除100条K线，这样可以避免日线这种大周期K线每天清理
						);
					}
					else {
						ZQDBReserveData(h, cycle
							, MaxValueCount + std::max<>(MaxTimePoint, (size_t)100) //至少多预留100条K线，这样可以避免日线这种大周期K线每天清理
						);
					}
				}
			}
		}
		if (CurValueCount > 0) {
			zqdb::Msg msg(ZQDB_MSG_UNKNOWN);
			build_data(h, cycle, msg);
			size_t data_sz = 0;
			auto data = msg.GetData(&data_sz);
			if (data && data_sz) {
				return save_data(h, cycle, msg);
			}
		}
	}
	return 0;
}

void MDMgr::close_record()
{
	if (record_fs_) {
		delete record_fs_;
		record_fs_ = nullptr;
	}
}

void MDMgr::save_record(HZQDB h, MDB_FIELD* diff_field, size_t diff_field_num, CTPCODEINFO* data)
{
	if (data) {
		//内部回调
		if (record_fs_) {
			try {
				zqdb::SerializeDiffDataToCSV(*record_fs_
					, MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE)
					, diff_field, diff_field_num, (const char*)data, true);
			}
			catch (...) {

			}
		}
		//std::lock_guard<std::mutex> lock(record_mt_);
		//record_infos_.emplace_back(h, *data, diff_field, diff_field_num);
	}
	else {
		//外部调用
		service().post([this] {
			if (!record_fs_) {
				if (normal_flag_ && ZQDBIsRecording()) {
					//正常情况下才初始化录制
					auto dir = ZQDBGetDataDir();
					if (!dir || !dir[0]) {
						return;
					}
					zqdb::AllExchange allexchange(g_ctp_module);
					for (auto hexchange : allexchange)
					{
						zqdb::AllProduct allproduct(hexchange);
						zqdb::Msg msg_exchange(ZQDB_MSG_UNKNOWN);
						zqdb::BuildStdTableMsg(msg_exchange
							, MDB_FIELD_TABLE(ZQDB, PRODUCT)
							, MDB_FIELD_TABLE_FIELD_COUNT(ZQDB, PRODUCT)
							, allproduct.data(), allproduct.size());
						save_data(hexchange, CYC_MAX, msg_exchange);
						for (auto hproduct : allproduct)
						{
							zqdb::AllCode allcode(hproduct);
							zqdb::Msg msg_product(ZQDB_MSG_UNKNOWN);
							zqdb::BuildStdTableMsg(msg_product
								, MDB_FIELD_TABLE(CTP, CODE)
								, MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE)
								, allcode.data(), allcode.size());
							save_data(hproduct, CYC_MAX, msg_product);
							for (auto hcode : allcode)
							{
								for (size_t cycle = 0; cycle < CYC_MAX; cycle++)
								{
									if (!ZQDBIsBaseCycle((PERIODTYPE)cycle)) {
										continue;
									}
									zqdb::Msg msg_cycle(ZQDB_MSG_UNKNOWN);
									build_data(hcode, (PERIODTYPE)cycle, msg_cycle);
									if (msg_cycle.GetParamAsUInt("elem_num")) {
										//只保存有效数据
										save_data(hcode, (PERIODTYPE)cycle, msg_cycle);
									}
								}
							}
						}
					}
					char buf[1024] = { 0 };
					snprintf(buf, 1024, "%s/%s/market/%d/market.csv", dir, STR_MODULE_NAME, tradingday_);
					try {
						record_fs_ = new std::fstream(buf, std::fstream::out /*| std::fstream::binary*/ | std::fstream::trunc);
						zqdb::SerializeTableTitleToCSV(*record_fs_
							, g_ctp_module, "CODE"
							, MDB_FIELD_TABLE(CTP, CODE)
							, MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE));
					}
					catch (...) {

					}
				}
			}
			else {
				if (ZQDBIsRecording()) {
					//正常录制
					try {
						record_fs_->flush();
					}
					catch (...) {

					}
				}
				else {
					//停止录制
					try {
						delete record_fs_;
					}
					catch (...) {

					};
					record_fs_ = nullptr;
				}
			}
		});
	}
}

void MDMgr::test_record(uint32_t& date, uint32_t& time)
{
	ASSERT(ZQDBIsTest());
	ASSERT(ZQDBIsTesting());
	std::string line;
	if (!record_fs_) {
		//初始化回测
		auto dir = ZQDBGetDataDir();
		if (!dir || !dir[0]) {
			return;
		}
		char buf[1024] = { 0 };
		snprintf(buf, 1024, "%s/%s/market/%d/market.csv", dir, STR_MODULE_NAME, tradingday_);
		try {
			record_fs_ = new std::fstream(buf, std::fstream::in /*| std::fstream::binary*/);
			std::getline(*record_fs_, line); //跳过头
		}
		catch (...) {

		}
	}
	//
	if (record_info_.h) {
		if (record_info_.data.Date < date
			|| (record_info_.data.Date == date && record_info_.data.Time <= time)) {
			service().post([this, record_info = record_info_] {
				ZQDBUpdateMarketData(record_info.h, (MDB_FIELD*)record_info.diff, record_info.diff_num, (CODEINFO*)&record_info.data, 0);
			});
			record_info_.clear();
		}
		else {
			date = record_info_.data.Date;
			time = record_info_.data.Time;
			return;
		}
	}
	while (zqdb::BuildDiffDataFromCSV(*record_fs_
		, MDB_FIELD_TABLE(CTP, CODE)
		, record_info_.diff, record_info_.diff_num, (char*)&record_info_.data, true)) {
		record_info_.h = ZQDBGetCode(record_info_.data.Code, record_info_.data.Exchange);
		if (record_info_.h) {
			zqdb::Code code(record_info_.h);
			if (!record_info_.data.Date)
				record_info_.data.Date = code->Date;
			if (record_info_.data.Date < date
				|| (record_info_.data.Date == date && record_info_.data.Time <= time)) {
				service().post([this, record_info = record_info_] {
					ZQDBUpdateMarketData(record_info.h, (MDB_FIELD*)record_info.diff, record_info.diff_num, (CODEINFO*)&record_info.data, 0);
				});
				record_info_.clear();
			}
			else {
				date = record_info_.data.Date;
				time = record_info_.data.Time;
				return;
			}
		}
	}
}

void MDMgr::set_offline(std::shared_ptr<CTPMarket> market)
{
	bool online_flag = false;
	for (auto market : markets_) {
		if (market->is_normal()) {
			online_flag = true;
			break;
		}
	}
	on_online(online_flag);
}

void MDMgr::set_online(std::shared_ptr<CTPMarket> market)
{
	bool online_flag = false;
	for (auto market : markets_) {
		if (market->is_normal()) {
			online_flag = true;
			break;
		}
	}
	on_online(online_flag);
}

void MDMgr::on_online(bool online_flag)
{
	online_flag_ = online_flag;

	if (init_flag_) {
		return;
	}

	if (online_flag_) {
		if (ZQDBIsTest()) {
			//
		}
		else {
			if (markets_.empty()) {
				//使用官方数据
				/*先不清理，保持数据连续
				//先清理数据
				zqdb::AllCode allcode(g_tora_module);
				for (auto h : allcode)
				{
					ZQDBRemoveData(h);
				}*/
				//重新订阅
				service().post(boost::bind(&This::subscribe, this));
			}
		}
		for (auto& exchange : all_exchange)
		{
			ZQDBSetStatus(exchange, ZQDB_STATUS_LOGGED);
		}
	}
	else {
		for (auto& exchange : all_exchange)
		{
			ZQDBSetStatus(exchange, is_stop() ? ZQDB_STATUS_OFFLINE : ZQDB_STATUS_CONNECT);
		}
	}
}

void MDMgr::begin_new_tradingday()
{
	normal_flag_ = false;

	for (auto& exchange : all_exchange)
	{
		ZQDBBeginNewTradingDay(exchange, /*old_normal ? SAVE_MARKET_DATA : */0);
	}
}

void MDMgr::clear_new_tradingday()
{
	boost::system::error_code ec;
	close_record(); //如果录制中，先停止录制，稍后会重新录制
	auto dara_dir = ZQDBGetDataDir();
	if (dara_dir && dara_dir[0]) {
		char buf[1024] = { 0 };
		snprintf(buf, 1024, "%s/%s/market", dara_dir, STR_MODULE_NAME);
		std::vector<boost::filesystem::path> old_dirs;
		boost::filesystem::directory_iterator end;
		for (boost::filesystem::directory_iterator pos(buf); pos != end; ++pos) {
			boost::filesystem::path sub_path(*pos);
			if (boost::filesystem::is_regular_file(sub_path)) {
				//
			}
			else if (boost::filesystem::is_directory(sub_path)) {
				size_t date = XUtil::stoi(sub_path.filename().string());
				if (date > 20230412 && date < tradingday_) {
					old_dirs.emplace_back(sub_path);
				}
			}
		}
		const size_t reserve = 0; //保留最近天数
		if (old_dirs.size() > reserve) {
			std::sort(old_dirs.begin(), old_dirs.end());
			for (size_t i = 0, j = old_dirs.size() - reserve; i < j; i++)
			{
				boost::filesystem::remove_all(old_dirs[i], ec);
			}
		}
	}

	//清理市场（市场、产品、代码、K线、明细等...）
	for (auto& exchange : all_exchange)
	{
		ZQDBDoNewTradingDayClear(exchange, CLEAR_MARKET_DATA);
		//自己实现清理数据
		/*zqdb::AllCode all_code(ZQDBGetExchange(pr.second->Exchange));
		for (size_t i = 0, j = all_code.size(); i < j; i++)
		{
		zqdb::Code code(all_code[i]);
		for (auto cycle = 0; cycle < CYC_MAX; cycle++)
		{
		ZQDBClearHisData(code, (PERIODTYPE)cycle, CLEAR_HIS_DATA);
		//GetOrOpenData(code_info.get(), (PERIODTYPE)cycle);
		//auto h = code.getdat
		}
		for (auto& tb : code_info->tb_data)
		{
		if (tb) {
		tb.RemoveValue(0, tb.GetValueCount());
		}
		}
		}*/
	}
}

std::map<const char*, std::shared_ptr<ExchangeInfo>, XUtil::strless> MDMgr::init_base(uint32_t trading_day)
{
	std::map<const char*, std::shared_ptr<ExchangeInfo>, XUtil::strless> exchanges;
	
	//中金所
	{
		auto exchange = std::make_shared<ExchangeInfo>();
		strcpy(exchange->Exchange, EXCHANGE_CFFEX);
		strcpy(exchange->Name, u8"中金所");
		exchange->TradeDay = trading_day;
		exchanges[exchange->Exchange] = exchange;
	}

	//上期所
	{
		auto exchange = std::make_shared<ExchangeInfo>();
		strcpy(exchange->Exchange, EXCHANGE_SHFE);
		strcpy(exchange->Name, u8"上期所");
		exchange->TradeDay = trading_day;
		exchanges[exchange->Exchange] = exchange;
	}

	//郑商所
	{
		auto exchange = std::make_shared<ExchangeInfo>();
		strcpy(exchange->Exchange, EXCHANGE_CZCE);
		strcpy(exchange->Name, u8"郑商所");
		exchange->TradeDay = trading_day;
		exchanges[exchange->Exchange] = exchange;
	}

	//大商所
	{
		auto exchange = std::make_shared<ExchangeInfo>();
		strcpy(exchange->Exchange, EXCHANGE_DCE);
		strcpy(exchange->Name, u8"大商所");
		exchange->TradeDay = trading_day;
		exchanges[exchange->Exchange] = exchange;
	}

	//上能源
	{
		auto exchange = std::make_shared<ExchangeInfo>();
		strcpy(exchange->Exchange, EXCHANGE_INE);
		strcpy(exchange->Name, u8"上能源");
		exchange->TradeDay = trading_day;
		exchanges[exchange->Exchange] = exchange;
	}

	//广期所
	{
		auto exchange = std::make_shared<ExchangeInfo>();
		strcpy(exchange->Exchange, EXCHANGE_GFEX);
		strcpy(exchange->Name, u8"广期所");
		exchange->TradeDay = trading_day;
		exchanges[exchange->Exchange] = exchange;
	}

	return std::move(exchanges);
}

void MDMgr::init_base_data()
{
	auto exchanges = init_base(tradingday_);
	for (auto& str : all_exchange)
	{
		auto it = exchanges.find(str);
		if (it == exchanges.end()) {
			continue;
		}
		auto& exchange = it->second;
		ASSERT(exchange);
		exchange->h = ZQDBUpdateOrInitExchange(exchange.get());
		ZQDBUpdateOrInitProductTable(MDB_FIELD_TABLE(ZQDB, PRODUCT), MDB_FIELD_TABLE_FIELD_COUNT(ZQDB, PRODUCT), exchange->h, MaxProductCountPerExchange);
		//
		std::string data;
		if (0 == load_data(exchange->h, CYC_MAX, data)) {
			init_data(exchange->h, CYC_MAX, data.empty() ? zqdb::Msg() : zqdb::Msg(data.data(), data.size()));
		}
		else if (ZQDBIsTest()) {
			//
		}
		else if (ZQDBIsRPC()) {
			zqdb::Msg msg(ZQDB_MSG_REQUEST_QUERY);
			msg.SetReqID(com::zqdb::proto::msg::MSG_REQUEST_QUERY_MD_CODE);
			msg.SetParam("Module", STR_MODULE_NAME);
			msg.SetParam("Exchange", exchange->Exchange);
			HNMSG hrsp = nullptr;
			ZQDBRequest(msg, &hrsp, 5000);
			if (hrsp) {
				zqdb::Msg rsp(hrsp, zqdb::Msg::AutoDelete);
				init_data(exchange->h, CYC_MAX, rsp, nullptr);
				save_data(exchange->h, CYC_MAX, rsp);
			}
		}
	}
	for (auto& pr : exchanges)
	{
		auto exchange = pr.second;
		ASSERT(exchange);
		//
		zqdb::AllProduct allproduct(exchange->h);
		for (auto hproduct : allproduct)
		{
			std::string data;
			if (0 == load_data(hproduct, CYC_MAX, data)) {
				init_data(hproduct, CYC_MAX, data.empty() ? zqdb::Msg() : zqdb::Msg(data.data(), data.size()));
			}
			else if (ZQDBIsTest()) {
				//
			}
			else if (ZQDBIsRPC()) {
				zqdb::Product product(hproduct);
				zqdb::Msg msg(ZQDB_MSG_REQUEST_QUERY);
				msg.SetReqID(com::zqdb::proto::msg::MSG_REQUEST_QUERY_MD_CODE);
				msg.SetParam("Module", STR_MODULE_NAME);
				msg.SetParam("Exchange", product->Exchange);
				msg.SetParam("Product", product->Product);
				HNMSG hrsp = nullptr;
				ZQDBRequest(msg, &hrsp, 5000);
				if (hrsp) {
					zqdb::Msg rsp(hrsp, zqdb::Msg::AutoDelete);
					init_data(hproduct, CYC_MAX, rsp, nullptr);
					save_data(hproduct, CYC_MAX, rsp);
				}
			}
			if (sub_all_ || ZQDBIsTest()) {
				//全推送需要使用者注意使用，应该是在盘后重启程序，如果使用持久化DB则应该启动前清理DB
				zqdb::AllCode allcode(hproduct);
				for (auto hcode : allcode)
				{
					if (ZQDBIsAnyDataExist(hcode)) {
						continue;
					}
					//初始化明细和更新第一条K线数据到DB
					{
						init_data(hcode, CYC_TICK, zqdb::Msg());
					}
					for (size_t cycle = 0; cycle < CYC_MAX; cycle++)
					{
						if (!ZQDBIsBaseCycle((PERIODTYPE)cycle)) {
							continue;
						}
						std::string data;
						if (exist_data(hcode, (PERIODTYPE)cycle)) {
							load_data(hcode, (PERIODTYPE)cycle, data);
						}
						if (!data.empty()) {
							//存在则初始化
							init_data(hcode, (PERIODTYPE)cycle, zqdb::Msg(data.data(), data.size()));
						}
						else if (ZQDBIsTest()) {
							//
						}
						else if (ZQDBIsRPC()) {
							//不存在就请求
							zqdb::Code code(hcode);
							zqdb::Msg msg(ZQDB_MSG_REQUEST_QUERY);
							msg.SetReqID(com::zqdb::proto::msg::MSG_REQUEST_QUERY_MD_DATA);
							msg.SetParam("Module", STR_MODULE_NAME);
							msg.SetParam("Exchange", code->Exchange);
							msg.SetParam("Code", code->Code);
							msg.SetParamAsInt("Period", cycle);
							HNMSG hrsp = nullptr;
							ZQDBRequest(msg, &hrsp, 5000);
							if (hrsp) {
								zqdb::Msg rsp(hrsp, zqdb::Msg::AutoDelete);
								init_data(hcode, (PERIODTYPE)cycle, rsp, nullptr);
								save_data(hcode, (PERIODTYPE)cycle, rsp);
							}
						}
					}
				}
			}
		}
	}
}

void update_market_data(CTPCODEINFO& new_code, const CThostFtdcDepthMarketDataField& data)
{
	///上次结算价
	new_code.YSettle = data.PreSettlementPrice;
	///昨收盘
	new_code.YClose = data.PreClosePrice;
	///本次结算价
	new_code.Settle = data.SettlementPrice;
	///涨停板价
	new_code.Upper = data.UpperLimitPrice;
	///跌停板价
	new_code.Lower = data.LowerLimitPrice;

	///昨持仓量
	new_code.PreOpenInterest = data.PreOpenInterest;
	///持仓量
	new_code.OpenInterest = data.OpenInterest;

	///昨虚实度
	new_code.PreDelta = data.PreDelta;
	///今虚实度
	new_code.CurrDelta = data.CurrDelta;

	///申买价一
	new_code.Bid1 = data.BidPrice1;
	///申买量一
	new_code.BidV1 = data.BidVolume1;
	///申卖价一
	new_code.Ask1 = data.AskPrice1;
	///申卖量一
	new_code.AskV1 = data.AskVolume1;
	///申买价二
	new_code.Bid2 = data.BidPrice2;
	///申买量二
	new_code.BidV2 = data.BidVolume2;
	///申卖价二
	new_code.Ask2 = data.AskPrice2;
	///申卖量二
	new_code.AskV2 = data.AskVolume2;
	///申买价三
	new_code.Bid3 = data.BidPrice3;
	///申买量三
	new_code.BidV3 = data.BidVolume3;
	///申卖价三
	new_code.Ask3 = data.AskPrice3;
	///申卖量三
	new_code.AskV3 = data.AskVolume3;
	///申买价四
	new_code.Bid4 = data.BidPrice4;
	///申买量四
	new_code.BidV4 = data.BidVolume4;
	///申卖价四
	new_code.Ask4 = data.AskPrice4;
	///申卖量四
	new_code.AskV4 = data.AskVolume4;
	///申买价五
	new_code.Bid5 = data.BidPrice5;
	///申买量五
	new_code.BidV5 = data.BidVolume5;
	///申卖价五
	new_code.Ask5 = data.AskPrice5;
	///申卖量五
	new_code.AskV5 = data.AskVolume5;
}

void update_market_data(CTPCODEINFO& new_code, uint32_t Date, uint32_t TimeSSS, const CThostFtdcDepthMarketDataField& data)
{
	new_code.Date = Date;
	new_code.Time = TimeSSS;
	//
	update_market_data(new_code, data);
}

void MDMgr::init_new_tradingday()
{
	if (ZQDBIsTest()) {

	}
	else {
		LOG4I(STR_MODULE_NAME " market start update market data to db");
		boost::system::error_code ec;
		auto dir = ZQDBGetDataDir();
		char buf[1024] = { 0 };
		snprintf(buf, 1024, "%s/%s/market/%d", dir, STR_MODULE_NAME
			, tradingday_);
		if (!boost::filesystem::exists(buf, ec)) {
			boost::filesystem::create_directories(buf, ec);
		}
		if (markets_.empty()) {
			init_base_data();
		}
		else {
			//更新数据
			std::map<const char*, std::shared_ptr<ExchangeInfo>, XUtil::strless> exchanges_;
			std::map<const char*, std::shared_ptr<ProductInfo>, XUtil::strless> products_;
			std::map<const char*, std::shared_ptr<CodeInfo>, XUtil::strless> codes_;
			for (auto& pr : init_exchanges_)
			{
				auto exchange = std::make_shared<ExchangeInfo>();
				strcpy(exchange->Exchange, pr.second->Exchange);
				strcpy(exchange->Name, gbk_to_utf8(pr.second->Name).c_str());
				ASSERT(exchanges_.find(exchange->Exchange) == exchanges_.end());
				exchanges_[exchange->Exchange] = exchange;
				exchange->sp = pr.second;
			}
			for (auto& pr : exchanges_)
			{
				auto exchange = pr.second;
				ASSERT(exchange);
				if (!exchange->h) {
					//第一次从DB初始化市场信息，接下来和最新市场信息比较
					exchange->h = ZQDBGetExchange(exchange->Exchange);
				}
			}
			for (auto& pr : init_products_)
			{
				auto product = std::make_shared<ProductInfo>();
				strcpy(product->Exchange, pr.second->Exchange);
				strcpy(product->Product, pr.second->Product);
				strcpy(product->Name, gbk_to_utf8(pr.second->Name).c_str());
				product->Type = pr.second->ProductClass;
				product->TradingUnit = pr.second->MinLimitOrderVolume;
				product->VolumeMultiple = pr.second->VolumeMultiple;
				ASSERT(products_.find(product->Product) == products_.end());
				products_[product->Product] = product;
				product->sp = pr.second;
			}
			for (auto& pr : products_)
			{
				auto product = pr.second;
				ASSERT(product);
				auto it = exchanges_.find(product->Exchange);
				if (it != exchanges_.end()) {
					product->parent = it->second;
				}
				if (!product->h) {
					product->h = ZQDBGetProduct(product->Product, product->Exchange);
				}
			}
			std::shared_ptr<CTPCommodityInfoEx> ctp_cur_commodity;
			std::shared_ptr<CTPCommodityInfoEx> ctp_prev_commodity;
			for (auto& pr : init_commoditys_)
			{
				std::shared_ptr<CTPCommodityInfoEx> ctp_commodity = pr.second;
				if (!ctp_cur_commodity || ctp_cur_commodity->product_ != ctp_commodity->product_) {
					ctp_cur_commodity = ctp_commodity;
					ctp_prev_commodity = nullptr;
				}

				if (ctp_prev_commodity) {
					ctp_commodity->pos = ctp_prev_commodity->pos + 1;
				}
				else {
					ctp_cur_commodity->pos = 0;
				}

				//原始合约
				auto code = std::make_shared<CodeInfo>();
				strcpy(code->Exchange, ctp_commodity->Exchange);
				strcpy(code->Product, ctp_commodity->Product);
				Instrument2Code(tradingday_, ctp_commodity, code->Code);
				//strcpy(code->Code, ctp_commodity->Code);
				//strcpy(code->Name, gbk_to_utf8(pr.second->Name).c_str());
				snprintf(code->Name, MAX_NAME_LENGTH, u8"%s", gbk_to_utf8(pr.second->Name).c_str());
				ASSERT(codes_.find(code->Code) == codes_.end());
				codes_[code->Code] = code;
				code->sp = ctp_commodity;

				/*//连续合约
				auto codeEx = std::make_shared<CodeInfo>();
				strcpy(codeEx->Exchange, ctp_commodity->Exchange);
				strcpy(codeEx->Product, ctp_commodity->Product);
				Instrument2CodeEx(ctp_commodity, codeEx->Code);
				if (ctp_commodity->pos) {
				snprintf(codeEx->Name, MAX_NAME_LENGTH, u8"%s连续%02d", gbk_to_utf8(ctp_commodity->product_->Name).c_str(), ctp_commodity->pos);
				}
				else {
				snprintf(codeEx->Name, MAX_NAME_LENGTH, u8"%s连续", gbk_to_utf8(ctp_commodity->product_->Name).c_str());
				}
				ASSERT(codes_.find(codeEx->Code) == codes_.end());
				codes_[codeEx->Code] = codeEx;
				codeEx->sp = ctp_commodity;*/

				//主力合约
				if (IsInstrumentMain(ctp_commodity)) {
					auto codeMain = std::make_shared<CodeInfo>();
					strcpy(codeMain->Exchange, ctp_commodity->Exchange);
					strcpy(codeMain->Product, ctp_commodity->Product);
					Instrument2Mian(ctp_commodity, codeMain->Code);
					snprintf(codeMain->Name, MAX_NAME_LENGTH, u8"%s主力", gbk_to_utf8(ctp_commodity->product_->Name).c_str());
					ASSERT(codes_.find(codeMain->Code) == codes_.end());
					codes_[codeMain->Code] = codeMain;
					codeMain->sp = ctp_commodity;
				}

				//次主力合约
				/*if (IsInstrumentSecond(ctp_commodity)) {
					auto codeSecond = std::make_shared<CodeInfo>();
					strcpy(codeSecond->Exchange, ctp_commodity->Exchange);
					strcpy(codeSecond->Product, ctp_commodity->Product);
					Instrument2Second(ctp_commodity, codeSecond->Code);
					snprintf(codeSecond->Name, MAX_NAME_LENGTH, u8"%s次主力", gbk_to_utf8(ctp_commodity->product_->Name).c_str());
					ASSERT(codes_.find(codeSecond->Code) == codes_.end());
					codes_[codeSecond->Code] = codeSecond;
					codeSecond->sp = ctp_commodity;
				}*/

				ctp_prev_commodity = ctp_commodity;
			}
			for (auto& pr : codes_)
			{
				auto code = pr.second;
				ASSERT(code);
				auto it = products_.find(code->Product);
				if (it != products_.end()) {
					code->parent = it->second;
				}
				if (!code->h) {
					code->h = ZQDBGetCode(code->Code, code->Exchange);
				}
			}

			/*const auto& rsp_logininfo = trader->rsp_logininfo();
			auto Date = 0;
			auto Time = CTPTime2Time(rsp_logininfo.LoginTime);
			for (auto& pr : codes_)
			{
				auto& code = pr.second;
				auto code_Date = CTPDate2Date(code->sp->ActionDay);
				if (Date < code_Date) {
					Date = code_Date;
				}
			}*/
			/*exchanges_[EXCHANGE_CFFEX]->Time = CTPTime2Time(rsp_logininfo.FFEXTime);
			exchanges_[EXCHANGE_SHFE]->Time = CTPTime2Time(rsp_logininfo.SHFETime);
			exchanges_[EXCHANGE_DCE]->Time = CTPTime2Time(rsp_logininfo.DCETime);
			exchanges_[EXCHANGE_CZCE]->Time = CTPTime2Time(rsp_logininfo.CZCETime);
			exchanges_[EXCHANGE_INE]->Time = CTPTime2Time(rsp_logininfo.IZQDBime);
			for (auto& pr : exchanges_)
			{
			auto& exchange = pr.second;
			if (!exchange->Time) {
			//
			}
			else if (Time < exchange->Time) {
			Time = exchange->Time; //时间用最最新时间
			}
			}*/
			for (auto& pr : exchanges_)
			{
				auto& exchange = pr.second;
				exchange->TradeDay = tradingday_;
				//exchange->Date = Date;//XUtil::NowDateTime(&exchange->Time);
				//if (!exchange->Time) {
				//	exchange->Time = Time;
				//}
				/*auto tTime = Time;
				if (strcmp(exchange->Exchange, EXCHANGE_CFFEX) == 0) {
					tTime = CTPTime2Time(rsp_logininfo.FFEXTime);
				}
				else if (strcmp(exchange->Exchange, EXCHANGE_SHFE) == 0) {
					tTime = CTPTime2Time(rsp_logininfo.SHFETime);
				}
				else if (strcmp(exchange->Exchange, EXCHANGE_DCE) == 0) {
					tTime = CTPTime2Time(rsp_logininfo.DCETime);
				}
				else if (strcmp(exchange->Exchange, EXCHANGE_CZCE) == 0) {
					tTime = CTPTime2Time(rsp_logininfo.CZCETime);
				}
				else if (strcmp(exchange->Exchange, EXCHANGE_INE) == 0) {
					tTime = CTPTime2Time(rsp_logininfo.INETime);
				}
				auto exchange_time = std::chrono::system_clock::from_time_t(XUtil::make_time_t(
					XUtil::GetYear(Date), XUtil::GetMonth(Date), XUtil::GetDay(Date)
					, XUtil::GetHour(tTime), XUtil::GetMinute(tTime), XUtil::GetSecond(tTime)));*/
				exchange->DiffSecs = 0;// std::chrono::duration_cast<std::chrono::seconds>(exchange_time - trader->login_time()).count();
				exchange->Status = exchange->sp->InstrumentStatus;
				exchange->EnterReason = exchange->sp->EnterReason;
			}
			for (auto& pr : products_)
			{
				auto& product = pr.second;
				product->Status = product->sp->InstrumentStatus;
				product->EnterReason = product->sp->EnterReason;
			}
			//再更新DB数据
			LOG4I(STR_MODULE_NAME " market update exchange to db...");
			for (auto& str : all_exchange)
			{
				auto it = exchanges_.find(str);
				if (it == exchanges_.end()) {
					continue;
				}
				auto& exchange = it->second;
				exchange->h = ZQDBUpdateOrInitExchange(exchange.get());
				ZQDBUpdateOrInitProductTable(MDB_FIELD_TABLE(ZQDB, PRODUCT), MDB_FIELD_TABLE_FIELD_COUNT(ZQDB, PRODUCT), exchange->h, MaxProductCountPerExchange);
			}
			LOG4I(STR_MODULE_NAME " market update product to db...");
			for (auto& pr : products_)
			{
				auto& product = pr.second;
				product->h = ZQDBUpdateOrInitProduct(product->parent->h, product.get());
				ZQDBUpdateOrInitCodeTable(MDB_FIELD_TABLE(CTP, CODE), MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE), product->h, MaxCodeCountPerProduct);
			}
			LOG4I(STR_MODULE_NAME " market update code to db...");
			for (auto& pr : codes_)
			{
				//这里初始化静态信息，行情数据等行情推送更新
				auto& code = pr.second;
				auto& product = code->parent;
				product->Type = code->sp->ProductClass;
				strncpy(code->TradeCode, code->sp->Code, MAX_CODE_LENGTH);
				//
				update_market_data(*code, *code->sp);
				///交割年份
				code->DeliveryYear = code->sp->DeliveryYear;
				///交割月
				code->DeliveryMonth = code->sp->DeliveryMonth;
				///市价单最大下单量
				code->MaxMarketOrderVolume = code->sp->MaxMarketOrderVolume;
				///市价单最小下单量
				code->MinMarketOrderVolume = code->sp->MinMarketOrderVolume;
				///限价单最大下单量
				code->MaxLimitOrderVolume = code->sp->MaxLimitOrderVolume;
				///限价单最小下单量
				code->MinLimitOrderVolume = code->sp->MinLimitOrderVolume;
				///合约数量乘数
				code->VolumeMultiple = code->sp->VolumeMultiple;
				///最小变动价位
				code->PriceTick = code->sp->PriceTick;
				///创建日
				code->CreateDate = CTPDate2Date(code->sp->CreateDate);
				///上市日
				code->OpenDate = CTPDate2Date(code->sp->OpenDate);
				///到期日
				code->ExpireDate = CTPDate2Date(code->sp->ExpireDate);
				///开始交割日
				code->StartDelivDate = CTPDate2Date(code->sp->StartDelivDate);
				///结束交割日
				code->EndDelivDate = CTPDate2Date(code->sp->EndDelivDate);
				///合约生命周期状态
				code->InstLifePhase = code->sp->InstLifePhase;
				///当前是否交易
				code->IsTrading = code->sp->IsTrading;
				///持仓类型
				code->PositionType = code->sp->PositionType;
				///持仓日期类型
				code->PositionDateType = code->sp->PositionDateType;
				///多头保证金率
				code->LongMarginRatio = code->sp->LongMarginRatio;
				///空头保证金率
				code->ShortMarginRatio = code->sp->ShortMarginRatio;
				///是否使用大额单边保证金算法
				code->MaxMarginSideAlgorithm = code->sp->MaxMarginSideAlgorithm;
				///基础商品代码
				strcpy(code->UnderlyingInstrID, code->sp->UnderlyingInstrID);
				///执行价
				code->StrikePrice = code->sp->StrikePrice;
				///期权类型
				code->OptionsType = code->sp->OptionsType;
				///合约基础商品乘数
				code->UnderlyingMultiple = code->sp->UnderlyingMultiple;
				///组合类型
				code->CombinationType = code->sp->CombinationType;

				bool need_init = false;
				if (!code->h) {
					need_init = true;
				}
				code->h = ZQDBUpdateOrInitCode(code->parent->h, (CODEINFO*)code.get());

				if (need_init) {
					if (sub_all_) {
						need_init = true;
					}
					else {
						need_init = false;
					}
				}

				bool init_flag[CYC_MAX] = { 0 };
				init_flag[CYC_TICK] = need_init;
				for (size_t cycle = 0; cycle < CYC_MAX; cycle++)
				{
					if (ZQDBIsBaseCycle((PERIODTYPE)cycle)) {
						init_flag[cycle] = exist_data(code->h, (PERIODTYPE)cycle);
						if (!init_flag[cycle]) {
							init_flag[cycle] = need_init;
						}
					}
				}
				//if (need_init) 
				{
					//初始化明细和更新第一条K线数据到DB
					if (init_flag[CYC_TICK])
					{
						init_data(code->h, CYC_TICK, zqdb::Msg());
					}

					for (size_t cycle = 0; cycle < CYC_MAX; cycle++)
					{
						if (!ZQDBIsBaseCycle((PERIODTYPE)cycle)) {
							continue;
						}
						if (!init_flag[cycle]) {
							continue;
						}
						std::string data;
						load_data(code->h, (PERIODTYPE)cycle, data);
						init_data(code->h, (PERIODTYPE)cycle, data.empty() ? zqdb::Msg() : zqdb::Msg(data.data(), data.size()));
						/*if (data.empty()) {
							CTPKDATA kdata = { 0 };
							now_data(code->h, (PERIODTYPE)cycle, &kdata);
							ZQDBAddKData(code->h, (PERIODTYPE)cycle, (const KDATA*)&kdata, sizeof(kdata));
						}*/
					}
				}
			}
		}
		LOG4I(STR_MODULE_NAME " market complete update market to db");
	}
}

void MDMgr::end_new_tradingday()
{
	//完成市场清理
	for (auto& exchange : all_exchange)
	{
		ZQDBEndNewTradingDay(exchange, 0);
	}
	ASSERT(init_flag_);
	init_flag_ = false;
	on_normal(true);
}

void MDMgr::update_data(const CThostFtdcDepthMarketDataField& data)
{
	if (!normal_flag_) {
		return;
	}
	HZQDB hs[3] = { 0 };
	if (ZQDBGetCodeByTradeCode(g_ctp_module, data.InstrumentID, hs, 3)) {
		for (auto h : hs)
		{
			if (!h) {
				break;
			}
			update_data(h, data);
		}
	}
}

bool IsValidMarketData(const CThostFtdcDepthMarketDataField& data)
{
	//对于无法纠正的错误返回false
	if (data.Volume == 0) {
		if (IsInvalidValue(data.LastPrice)) {
			return false;
		}
	}
	else {
		if (IsInvalidValue(data.LastPrice)) {
			return false;
		}
		if ((data.OpenPrice - data.LowestPrice) < -ZQDB_ZERO_CTL_VALUE || (data.OpenPrice - data.HighestPrice) > ZQDB_ZERO_CTL_VALUE) {
			return false;
		}
		if ((data.LastPrice - data.LowestPrice) < -ZQDB_ZERO_CTL_VALUE || (data.LastPrice - data.HighestPrice) > ZQDB_ZERO_CTL_VALUE) {
			return false;
		}
	}
	return true;
}

bool MDMgr::update_data_price(CTPCODEINFO& code, const CThostFtdcDepthMarketDataField& data)
{
	if (data.Volume == 0) {
		/*code.Open = std::numeric_limits<double>::infinity();
		code.High = std::numeric_limits<double>::infinity();
		code.Low = std::numeric_limits<double>::infinity();
		code.Close = std::numeric_limits<double>::infinity();*/
		code.Open = data.LastPrice;
		code.High = data.LastPrice;
		code.Low = data.LastPrice;
		code.Close = data.LastPrice;
		code.Volume = 0;
		code.Amount = 0;
		return false;
	}
	else {
		if (IsInvalidValue(data.OpenPrice) || IsInvalidValue(data.LowestPrice) || IsInvalidValue(data.HighestPrice)) {
			code.Open = data.LastPrice;
			code.High = data.LastPrice;
			code.Low = data.LastPrice;
			code.Close = data.LastPrice;
		}
		else {
			if ((data.OpenPrice - data.LowestPrice) > -ZQDB_ZERO_CTL_VALUE && (data.OpenPrice - data.HighestPrice) < ZQDB_ZERO_CTL_VALUE) {
				code.Open = data.OpenPrice;
				code.High = data.HighestPrice;
				code.Low = data.LowestPrice;
			}
			else {
				code.Open = data.LastPrice;
				code.High = data.LastPrice;
				code.Low = data.LastPrice;
			}
			if ((data.ClosePrice - data.LowestPrice) > -ZQDB_ZERO_CTL_VALUE && (data.ClosePrice - data.HighestPrice) < ZQDB_ZERO_CTL_VALUE) {
				code.Close = data.ClosePrice;
			}
			else {
				code.Close = data.LastPrice;
			}
		}
		code.Volume = data.Volume;
		code.Amount = data.Turnover;
	}
	ASSERT(!IsInvalidValue(code.Close));
	ASSERT(!IsZeroValue(code.Close));
	return true;
}

void MDMgr::update_data(HZQDB h, const CThostFtdcDepthMarketDataField& data, const CThostFtdcInstrumentStatusField* status)
{
	bool sub = false;
	if (!sub_all_) {
		sub = ZQDBIsSubscribeMarketData(h);
		if (!sub) {
			return; //过滤取消订阅
		}
	}

#if 1
	uint32_t Date = 0, Time = 0, Millis = 0;
	Date = XUtil::NowDateTimeMillis(&Time, &Millis);
	auto TimeSSS = Time * 1000 + Millis;
#else
	auto Date = CTPDate2Date(data.ActionDay);
	auto Time = CTPTime2Time(data.UpdateTime); 
	auto TimeSSS = Time * 1000 + data.UpdateMillisec;
	if (!Date) {
		return;
	}
#endif
	//过滤周末测试行情
	size_t pos = 0;
	auto cmp = ZQDBIsNotInTradingTime(h, Date, Time, &pos);
	if (cmp < 0) {
		uint32_t date = 0, time = 0;
		time = ZQDBGetOpenTime(h, &date, pos);
		if (Date < date) {
			return;
		}
		if (XUtil::DiffTimeMinutes(Time, time) > 30) {
			return;
		}
	}
	else if (cmp > 0) {
		uint32_t date = 0, time = 0;
		time = ZQDBGetCloseTime(h, &date, pos);
		if (Date > date) {
			return;
		}
		if (XUtil::DiffTimeMinutes(time, Time) > 30) {
			return;
		}
	}

	zqdb::CodeT<CTPCODEINFO> obj(h);
	const auto& code = *(CTPCODEINFO*)obj;
	if (!IsValidMarketData(data)) {
		return;
	}

	auto new_code = code;
	bool bNewVolumeInValid = !update_data_price(new_code, data);
	if (!bNewVolumeInValid) {
		if (code.Volume > data.Volume || code.Amount > data.Turnover) {
			return;
		}
	}

	update_market_data(new_code, Date, TimeSSS, data);
	if (status) {
		new_code.EnterReason = status->EnterReason;
		new_code.Status = status->InstrumentStatus;
	}

	bool update = false;
	MDB_FIELD diff_field[MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE)] = { 0 };
	size_t diff_field_num = 0;
	if (!code.Date) {
		//第一次行情过来只更新快照，接下来再更新明细/K线等
		update = true;
		diff_field_num = zqdb::BuildFullFTDiffField(diff_field);
		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, YCLOSE);
		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, YSETTLE);
		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, UPPER);
		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, LOWER);
		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, PreOpenInterest);
		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, PreDelta);
		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, CurrDelta);
		ZQDBUpdateMarketData(h, diff_field, diff_field_num, (CTPCODEINFO*)&new_code, UPDATE_MARKET_DATA_FLAG_INIT);
	}
	else {
		diff_field_num = mdb::FieldOp::Diff(MDB_FIELD_TABLE(CTP, CODE), MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE), (const char*)&new_code, (const char*)&code, diff_field);
		for (size_t i = 0; i < diff_field_num; i++)
		{
			if (diff_field[i].index != MDB_FIELD_INDEX(CTP, CODE, DATE) && diff_field[i].index != MDB_FIELD_INDEX(CTP, CODE, TIME)) {
				update = true;
				break;
			}
		}
		if (update) {
			ZQDBUpdateMarketData(h, diff_field, diff_field_num, (CTPCODEINFO*)&new_code, 0);
		}
	}
	if (update) {
		if (ZQDBServer()) {
			push_flag_ = true;
			for (size_t i = 0; i < diff_field_num; i++)
			{
				pushs_[h][diff_field[i].index] = true;
			}
		}
		std::string strdiff;
		bool broadcast = ZQDBIPCIsAnySubscribe(&h, 1) || ZQDBRPCIsAnySubscribe(&h, 1);
		if (broadcast
#if USE_LOG_FILE
			|| ZQDBIsRecording()
#endif
			) {
			com::zqdb::proto::msg::RtnSubscribe rtn;
			auto one = rtn.add_codes();
			zqdb::BuildCodeInfo(one, &new_code, diff_field, diff_field_num);
			strdiff = rtn.SerializeAsString();
		}
		if (broadcast) {
			zqdb::Msg msg(ZQDB_MSG_NOTIFY);
			msg.SetReqID(com::zqdb::proto::msg::MSG_NOTIFY_MD_SUBSCRIBE);
			msg.SetParam("Module", STR_MODULE_NAME);
			msg.SetParam("Exchange", code.Exchange);
			msg.SetParam("Code", code.Code);
			struct BroadcastUserData {
				zqdb::Msg& msg;
				HZQDB h;
			}user_data{ msg, h };
			if (ZQDBIPCIsAnySubscribe(&h, 1)) {
				ZQDBBroadcastIPC([](HZQDB peer, void* user_data) -> bool {
					auto data = (BroadcastUserData*)user_data;
					if (ZQDBPeerIsAnySubscribe(peer, &data->h, 1)) {
						auto& msg = data->msg;
						msg.SetNode(peer);
						ZQDBSendMsg(msg);
					}
					return false;
				}, &user_data);
			}
			if (ZQDBRPCIsAnySubscribe(&h, 1)) {
				msg.SetData(strdiff.data(), strdiff.size());
				ZQDBBroadcastRPC([](HZQDB peer, void* user_data) -> bool {
					auto data = (BroadcastUserData*)user_data;
					if (ZQDBPeerIsAnySubscribe(peer, &data->h, 1)) {
						auto& msg = data->msg;
						msg.SetNode(peer);
						ZQDBSendMsg(msg);
					}
					return false;
				}, &user_data);
			}
		}
#if USE_LOG_FILE
		if (ZQDBIsRecording()) {
			tagZQDBRecordHead head;
			head.time = Time;
			head.type = RECORD_TYPE_UpdateMarketData;
			const tagZQDBRecordBuf buf[1] = {
				{ strdiff.data(), strdiff.size() },
			};
			ZQDBWriteRecordData(g_ctp_module, &head, buf, 1);
			}
#else
		if (ZQDBIsRecording()) {
			save_record(h, diff_field, diff_field_num, &new_code);
		}
#endif
	}
}

void MDMgr::update_msg(const com::zqdb::proto::msg::RtnSubscribe& msg)
{
	for (auto one : msg.codes())
	{
		auto h = ZQDBGetCode(one.code().c_str(), one.exchange().c_str());
		if (h) {
			bool sub = false;
			if (!sub_all_) {
				sub = ZQDBIsSubscribeMarketData(h);
				if (!sub) {
					continue; //过滤取消订阅
				}
			}

			zqdb::CodeT<CTPCODEINFO> obj(h);
			const auto& code = *(CTPCODEINFO*)obj;
			auto new_code = code;
			MDB_FIELD diff_field[MDB_FIELD_TABLE_FIELD_COUNT(CTP, CODE)] = { 0 };
			size_t diff_field_num = zqdb::UpdateCodeInfo(h, one, &new_code, diff_field);
			//if (diff_field_num > 0) 
			{
				if (!code.Date) {
					//第一次行情过来只更新快照，接下来再更新明细/K线等
					ZQDBUpdateMarketData(h, diff_field, diff_field_num, (CODEINFO*)&new_code, UPDATE_MARKET_DATA_FLAG_INIT);
				}
				else {
					ZQDBUpdateMarketData(h, diff_field, diff_field_num, (CODEINFO*)&new_code, 0);
				}
				if (ZQDBIsRecording()) {
					save_record(h, diff_field, diff_field_num, &new_code);
				}
			}
		}
	}
}

void MDMgr::update_status(const CThostFtdcInstrumentStatusField& status)
{
	/*if (status.InstrumentID[0]) {
	auto it = init_products_.find(status.InstrumentID);
	if (it != init_products_.end()) {
	update_status(nullptr, status.InstrumentID, nullptr, status);
	}
	else {
	auto it = init_commoditys_.find(status.InstrumentID);
	if (it != init_commoditys_.end()) {
	if (IsInstrumentMain(it->second)) {
	char Code[MAX_CODE_LENGTH + 1] = { 0 };
	Instrument2Mian(it->second, Code);
	update_status(nullptr, nullptr, Code, status);
	}
	{

	char Code[MAX_CODE_LENGTH + 1] = { 0 };
	Instrument2Code(it->second, Code);
	update_status(nullptr, nullptr, Code, status);
	}
	{
	char Code[MAX_CODE_LENGTH + 1] = { 0 };
	Instrument2CodeEx(it->second, Code);
	update_status(nullptr, nullptr, Code, status);
	}
	}
	}
	}
	else {
	auto it = init_exchanges_.find(status.ExchangeID);
	if (it != init_exchanges_.end()) {
	update_status(status.ExchangeID, nullptr, nullptr, status);
	}
	}*/
}

void MDMgr::update_status(const char* Exchange, const char* Product, const char* Code, const CThostFtdcInstrumentStatusField& status)
{
	//if (Code) {
	//	auto it = codes_.find(Code);
	//	if (it == codes_.end()) {
	//		return;
	//	}
	//	auto& info = it->second;
	//	CTPCODEINFO& code = *info;
	//	size_t diff_field_num = 0;
	//	MDB_FIELD diff_field[2] = { 0 };
	//	if (code.Status != status.InstrumentStatus) {
	//		code.Status = status.InstrumentStatus;
	//		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, STATUS);
	//	}
	//	if (code.EnterReason != status.EnterReason) {
	//		code.EnterReason = status.EnterReason;
	//		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(CTP, CODE, ENTER_REASON);
	//	}
	//	if (diff_field_num > 0) {
	//		ZQDBUpdateMarketData(info->h, diff_field, diff_field_num, &code);
	//	}
	//}
	//else if (Product) {
	//	auto it = products_.find(Product);
	//	if (it == products_.end()) {
	//		return;
	//	}
	//	auto& info = it->second;
	//	PRODUCTINFO& product = *info;
	//	size_t diff_field_num = 0;
	//	MDB_FIELD diff_field[2] = { 0 };
	//	if (product.Status != status.InstrumentStatus) {
	//		product.Status = status.InstrumentStatus;
	//		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(ZQDB, PRODUCT, STATUS);
	//	}
	//	if (product.EnterReason != status.EnterReason) {
	//		product.EnterReason = status.EnterReason;
	//		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(ZQDB, PRODUCT, ENTER_REASON);
	//	}
	//	if (diff_field_num > 0) {
	//		ZQDBUpdateMarketData(info->h, diff_field, diff_field_num, &product);

	//		//更新市场状态
	//		/*CThostFtdcInstrumentStatusField exchagne_status = status;
	//		bool update = true;
	//		if (product.Status == XMT_Status_BeforeTrading) {
	//		}
	//		else if (product.Status == XMT_Status_NoTrading || product.Status == XMT_Status_Closed) {
	//			for (auto& pr : products_) {
	//				if (strcmp(pr.second->Exchange, product.Exchange) == 0) {
	//					if (pr.second->Status != product.Status) {
	//						update = false;
	//						break;
	//					}
	//				}
	//			}
	//		}
	//		else {
	//			exchagne_status.InstrumentStatus = THOST_FTDC_IS_Continous;
	//		}
	//		if (update) {
	//			update_status(product.Exchange, nullptr, nullptr, exchagne_status);
	//		}*/
	//	}
	//}
	//else if (Exchange) {
	//	auto it = exchanges_.find(Exchange);
	//	if (it == exchanges_.end()) {
	//		return;
	//	}
	//	auto& info = it->second;
	//	EXCHANGEINFO& exchange = *info;
	//	size_t diff_field_num = 0;
	//	MDB_FIELD diff_field[2] = { 0 };
	//	if (exchange.Status != status.InstrumentStatus) {
	//		exchange.Status = status.InstrumentStatus;
	//		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(ZQDB, EXCHANGE, STATUS);
	//	}
	//	if (exchange.EnterReason != status.EnterReason) {
	//		exchange.EnterReason = status.EnterReason;
	//		diff_field[diff_field_num++] = MDB_FIELD_TABLE_FIELD(ZQDB, EXCHANGE, ENTER_REASON);
	//	}
	//	if (diff_field_num > 0) {
	//		ZQDBUpdateMarketData(info->h, diff_field, diff_field_num, &exchange);
	//	}
	//}
}
